feat(anthropic): emit streaming latency span attrs

packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/streaming.py

@@ -35,6 +35,58 @@
 
 logger = logging.getLogger(__name__)
 
+FR_STREAMING_TIME_TO_FIRST_TOKEN_MS = "fortifyroot.llm.streaming.time_to_first_token_ms"
+FR_STREAMING_TIME_TO_GENERATE_MS = "fortifyroot.llm.streaming.time_to_generate_ms"
+GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS = getattr(
+    SpanAttributes,
+    "GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS",
+    SpanAttributes.LLM_USAGE_CACHE_READ_INPUT_TOKENS,
+)
+GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS = getattr(
+    SpanAttributes,
+    "GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS",
+    SpanAttributes.LLM_USAGE_CACHE_CREATION_INPUT_TOKENS,
+)
+
+
+def _streaming_latency_ms(start_time, end_time):
+    return int(round((end_time - start_time) * 1000))
+
+
+def _is_streaming_token_item(item):
+    if getattr(item, "type", None) != "content_block_delta":
+        return False
+
+    delta = getattr(item, "delta", None)
+    delta_type = getattr(delta, "type", None)
+    if delta_type == "text_delta":
+        return bool(getattr(delta, "text", None))
+    if delta_type == "thinking_delta":
+        return bool(getattr(delta, "thinking", None))
+    if delta_type == "input_json_delta":
+        return bool(getattr(delta, "partial_json", None))
+    return False
+
+
+def _set_streaming_first_token_latency(span, start_time, first_token_time):
+    if not span or not span.is_recording():
+        return
+    set_span_attribute(
+        span,
+        FR_STREAMING_TIME_TO_FIRST_TOKEN_MS,
+        _streaming_latency_ms(start_time, first_token_time),
+    )
+
+
+def _set_streaming_time_to_generate(span, first_token_time, end_time):
+    if not first_token_time or not span or not span.is_recording():
+        return
+    set_span_attribute(
+        span,
+        FR_STREAMING_TIME_TO_GENERATE_MS,
+        _streaming_latency_ms(first_token_time, end_time),
+    )
+
 
 @dont_throw
 def _process_response_item(item, complete_response):
@@ -102,10 +154,10 @@ def _set_token_usage(
     set_span_attribute(span, SpanAttributes.LLM_USAGE_TOTAL_TOKENS, total_tokens)
 
     set_span_attribute(
-        span, SpanAttributes.GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS, cache_read_tokens
+        span, GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS, cache_read_tokens
     )
     set_span_attribute(
-        span, SpanAttributes.GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS, cache_creation_tokens
+        span, GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS, cache_creation_tokens
     )
 
     set_span_attribute(
@@ -182,6 +234,7 @@ def __init__(
 
         self._complete_response = {"events": [], "model": "", "usage": {}, "id": ""}
         self._instrumentation_completed = False
+        self._first_token_time = None
         self._pending_item = None  # FR: pending-item buffer for streaming safety
         self._streaming_safety = AnthropicStreamingSafety(span, getattr(span, "name", "anthropic.chat"))  # FR: streaming safety init
 
@@ -252,6 +305,13 @@ def __next__(self):  # FR: rewritten for pending-item safety buffer
                 raise e
 
             item = self._streaming_safety.process_item(item)  # FR: streaming safety processing
+            if self._first_token_time is None and _is_streaming_token_item(item):
+                self._first_token_time = time.time()
+                _set_streaming_first_token_latency(
+                    self._span,
+                    self._start_time,
+                    self._first_token_time,
+                )
             if self._pending_item is None:  # FR: pending-item buffer logic
                 self._pending_item = item  # FR: pending-item buffer logic
                 continue  # FR: pending-item buffer logic
@@ -275,8 +335,14 @@ def _handle_completion(self):
             GenAIAttributes.GEN_AI_RESPONSE_ID,
             self._complete_response.get("id"),
         )
+        end_time = time.time()
+        _set_streaming_time_to_generate(
+            self._span,
+            self._first_token_time,
+            end_time,
+        )
         if self._duration_histogram:
-            duration = time.time() - self._start_time
+            duration = end_time - self._start_time
             self._duration_histogram.record(
                 duration,
                 attributes=metric_attributes,
@@ -369,6 +435,7 @@ def __init__(
 
         self._complete_response = {"events": [], "model": "", "usage": {}, "id": ""}
         self._instrumentation_completed = False
+        self._first_token_time = None
         self._pending_item = None  # FR: pending-item buffer for streaming safety
         self._streaming_safety = AnthropicStreamingSafety(span, getattr(span, "name", "anthropic.chat"))  # FR: streaming safety init
 
@@ -442,6 +509,13 @@ async def __anext__(self):  # FR: rewritten for pending-item safety buffer
                 raise
 
             item = self._streaming_safety.process_item(item)  # FR: streaming safety processing
+            if self._first_token_time is None and _is_streaming_token_item(item):
+                self._first_token_time = time.time()
+                _set_streaming_first_token_latency(
+                    self._span,
+                    self._start_time,
+                    self._first_token_time,
+                )
             if self._pending_item is None:  # FR: pending-item buffer logic
                 self._pending_item = item  # FR: pending-item buffer logic
                 continue  # FR: pending-item buffer logic
@@ -463,9 +537,15 @@ def _complete_instrumentation(self):
             request_model=self._kwargs.get("model"),
         )
         set_span_attribute(self._span, GenAIAttributes.GEN_AI_RESPONSE_ID, self._complete_response.get("id"))
+        end_time = time.time()
+        _set_streaming_time_to_generate(
+            self._span,
+            self._first_token_time,
+            end_time,
+        )
 
         if self._duration_histogram:
-            duration = time.time() - self._start_time
+            duration = end_time - self._start_time
             self._duration_histogram.record(
                 duration,
                 attributes=metric_attributes,
@@ -482,6 +562,7 @@ def _complete_instrumentation(self):
                 if usage := self._complete_response.get("usage"):
                     completion_tokens = usage.get("output_tokens", 0)
                 else:
+                    completion_tokens = 0
                     completion_content = ""
                     if self._complete_response.get("events"):
                         model_name = self._complete_response.get("model") or None

packages/opentelemetry-instrumentation-anthropic/tests/test_safety_unit.py

@@ -3,12 +3,15 @@
 import pytest
 
 from opentelemetry.instrumentation.anthropic import safety
+from opentelemetry.instrumentation.anthropic.config import Config
 from opentelemetry.instrumentation.anthropic.streaming_safety import (
     AnthropicStreamingSafety,
 )
 from opentelemetry.instrumentation.anthropic.streaming import (
     AnthropicAsyncStream,
     AnthropicStream,
+    FR_STREAMING_TIME_TO_FIRST_TOKEN_MS,
+    FR_STREAMING_TIME_TO_GENERATE_MS,
 )
 from opentelemetry.instrumentation.fortifyroot import SafetyDecision, SafetyResult
 from opentelemetry.instrumentation.fortifyroot import (
@@ -410,6 +413,89 @@ def test_anthropic_sync_stream_span_not_ended_before_last_item_consumed():
     assert len(exporter.get_finished_spans()) == 1
 
 
+def test_anthropic_sync_stream_sets_streaming_latency_span_attrs(monkeypatch):
+    clear_safety_handlers()
+    exporter, tracer = _test_tracer()
+
+    timestamps = iter([101.25, 103.0])
+    monkeypatch.setattr(
+        "opentelemetry.instrumentation.anthropic.streaming.time.time",
+        lambda: next(timestamps),
+    )
+
+    span = tracer.start_span("anthropic.chat")
+    stream = AnthropicStream(
+        span,
+        _Iterator(
+            [
+                SimpleNamespace(
+                    type="content_block_start",
+                    index=0,
+                    content_block=SimpleNamespace(type="text"),
+                ),
+                SimpleNamespace(
+                    type="content_block_delta",
+                    index=0,
+                    delta=SimpleNamespace(type="text_delta", text="hello"),
+                ),
+            ]
+        ),
+        SimpleNamespace(count_tokens=lambda text: len(text)),
+        100.0,
+        kwargs={},
+    )
+
+    first = next(stream)
+    second = next(stream)
+    with pytest.raises(StopIteration):
+        next(stream)
+
+    finished = exporter.get_finished_spans()
+    assert first.type == "content_block_start"
+    assert second.type == "content_block_delta"
+    assert len(finished) == 1
+    assert finished[0].attributes[FR_STREAMING_TIME_TO_FIRST_TOKEN_MS] == 1250
+    assert finished[0].attributes[FR_STREAMING_TIME_TO_GENERATE_MS] == 1750
+
+
+def test_anthropic_sync_stream_without_token_omits_streaming_latency_attrs(monkeypatch):
+    clear_safety_handlers()
+    exporter, tracer = _test_tracer()
+
+    timestamps = iter([301.0])
+    monkeypatch.setattr(
+        "opentelemetry.instrumentation.anthropic.streaming.time.time",
+        lambda: next(timestamps),
+    )
+
+    span = tracer.start_span("anthropic.chat")
+    stream = AnthropicStream(
+        span,
+        _Iterator(
+            [
+                SimpleNamespace(
+                    type="content_block_start",
+                    index=0,
+                    content_block=SimpleNamespace(type="text"),
+                )
+            ]
+        ),
+        SimpleNamespace(count_tokens=lambda text: len(text)),
+        300.0,
+        kwargs={},
+    )
+
+    first = next(stream)
+    with pytest.raises(StopIteration):
+        next(stream)
+
+    finished = exporter.get_finished_spans()
+    assert first.type == "content_block_start"
+    assert len(finished) == 1
+    assert FR_STREAMING_TIME_TO_FIRST_TOKEN_MS not in finished[0].attributes
+    assert FR_STREAMING_TIME_TO_GENERATE_MS not in finished[0].attributes
+
+
 @pytest.mark.asyncio
 async def test_anthropic_async_stream_span_not_ended_before_last_item_consumed():
     """D-13: Async variant - span must not be ended until the consumer has processed the last pending item."""
@@ -444,3 +530,50 @@ async def test_anthropic_async_stream_span_not_ended_before_last_item_consumed()
 
     assert stream._instrumentation_completed
     assert len(exporter.get_finished_spans()) == 1
+
+
+@pytest.mark.asyncio
+async def test_anthropic_async_stream_sets_streaming_latency_span_attrs(monkeypatch):
+    clear_safety_handlers()
+    exporter, tracer = _test_tracer()
+    monkeypatch.setattr(Config, "enrich_token_usage", True)
+
+    timestamps = iter([201.1, 202.4])
+    monkeypatch.setattr(
+        "opentelemetry.instrumentation.anthropic.streaming.time.time",
+        lambda: next(timestamps),
+    )
+
+    span = tracer.start_span("anthropic.chat")
+    stream = AnthropicAsyncStream(
+        span,
+        _AsyncIterator(
+            [
+                SimpleNamespace(
+                    type="content_block_start",
+                    index=0,
+                    content_block=SimpleNamespace(type="text"),
+                ),
+                SimpleNamespace(
+                    type="content_block_delta",
+                    index=0,
+                    delta=SimpleNamespace(type="text_delta", text="hello"),
+                ),
+            ]
+        ),
+        SimpleNamespace(count_tokens=lambda text: len(text)),
+        200.0,
+        kwargs={},
+    )
+
+    first = await stream.__anext__()
+    second = await stream.__anext__()
+    with pytest.raises(StopAsyncIteration):
+        await stream.__anext__()
+
+    finished = exporter.get_finished_spans()
+    assert first.type == "content_block_start"
+    assert second.type == "content_block_delta"
+    assert len(finished) == 1
+    assert finished[0].attributes[FR_STREAMING_TIME_TO_FIRST_TOKEN_MS] == 1100
+    assert finished[0].attributes[FR_STREAMING_TIME_TO_GENERATE_MS] == 1300