fix(streaming): use monotonic clamped latency timings

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

@@ -556,7 +556,7 @@ def _wrap(
 
     kwargs = _apply_prompt_safety(span, kwargs, name)
     _handle_input(span, event_logger, kwargs)
-    start_time = time.time()
+    start_time = time.perf_counter()
     # ST-10.4 (review-driven 2026-05-17): make the anthropic.chat span
     # the AMBIENT OTel context for the duration of the SDK call so the
     # FortifyRoot retry handler (wrapping
@@ -575,19 +575,19 @@ def _wrap(
         with trace.use_span(span, end_on_exit=False):
             response = wrapped(*args, **kwargs)
     except Exception as e:  # pylint: disable=broad-except
-        end_time = time.time()
+        end_time = time.perf_counter()
         attributes = error_metrics_attributes(e)
 
         if duration_histogram:
-            duration = end_time - start_time
+            duration = max(0, end_time - start_time)
             duration_histogram.record(duration, attributes=attributes)
 
         if exception_counter:
             exception_counter.add(1, attributes=attributes)
 
         raise e
 
-    end_time = time.time()
+    end_time = time.perf_counter()
 
     if is_streaming_response(response):
         return AnthropicStream(
@@ -636,7 +636,7 @@ def _wrap(
             )
 
             if duration_histogram:
-                duration = time.time() - start_time
+                duration = max(0, time.perf_counter() - start_time)
                 duration_histogram.record(
                     duration,
                     attributes=metric_attributes,
@@ -697,18 +697,18 @@ async def _awrap(
     )
     kwargs = await asyncio.to_thread(_apply_prompt_safety, span, kwargs, name)  # FR: async safety
     await _ahandle_input(span, event_logger, kwargs)
-    start_time = time.time()
+    start_time = time.perf_counter()
     # ST-10.4 (review-driven 2026-05-17): see sync _wrap above for
     # rationale on use_span(end_on_exit=False) around the wrapped call.
     try:
         with trace.use_span(span, end_on_exit=False):
             response = await wrapped(*args, **kwargs)
     except Exception as e:  # pylint: disable=broad-except
-        end_time = time.time()
+        end_time = time.perf_counter()
         attributes = error_metrics_attributes(e)
 
         if duration_histogram:
-            duration = end_time - start_time
+            duration = max(0, end_time - start_time)
             duration_histogram.record(duration, attributes=attributes)
 
         if exception_counter:
@@ -766,7 +766,7 @@ async def _awrap(
         )
 
         if duration_histogram:
-            duration = time.time() - start_time
+            duration = max(0, time.perf_counter() - start_time)
             duration_histogram.record(
                 duration,
                 attributes=metric_attributes,

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

@@ -50,7 +50,7 @@
 
 
 def _streaming_latency_ms(start_time, end_time):
-    return int(round((end_time - start_time) * 1000))
+    return int(round(max(0, end_time - start_time) * 1000))
 
 
 def _is_streaming_token_item(item):
@@ -306,7 +306,7 @@ def __next__(self):  # FR: rewritten for pending-item safety buffer
 
             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()
+                self._first_token_time = time.perf_counter()
                 _set_streaming_first_token_latency(
                     self._span,
                     self._start_time,
@@ -335,14 +335,14 @@ def _handle_completion(self):
             GenAIAttributes.GEN_AI_RESPONSE_ID,
             self._complete_response.get("id"),
         )
-        end_time = time.time()
+        end_time = time.perf_counter()
         _set_streaming_time_to_generate(
             self._span,
             self._first_token_time,
             end_time,
         )
         if self._duration_histogram:
-            duration = end_time - self._start_time
+            duration = max(0, end_time - self._start_time)
             self._duration_histogram.record(
                 duration,
                 attributes=metric_attributes,
@@ -510,7 +510,7 @@ async def __anext__(self):  # FR: rewritten for pending-item safety buffer
 
             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()
+                self._first_token_time = time.perf_counter()
                 _set_streaming_first_token_latency(
                     self._span,
                     self._start_time,
@@ -537,15 +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()
+        end_time = time.perf_counter()
         _set_streaming_time_to_generate(
             self._span,
             self._first_token_time,
             end_time,
         )
 
         if self._duration_histogram:
-            duration = end_time - self._start_time
+            duration = max(0, end_time - self._start_time)
             self._duration_histogram.record(
                 duration,
                 attributes=metric_attributes,

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

@@ -12,6 +12,7 @@
     AnthropicStream,
     FR_STREAMING_TIME_TO_FIRST_TOKEN_MS,
     FR_STREAMING_TIME_TO_GENERATE_MS,
+    _streaming_latency_ms,
 )
 from opentelemetry.instrumentation.fortifyroot import SafetyDecision, SafetyResult
 from opentelemetry.instrumentation.fortifyroot import (
@@ -27,6 +28,10 @@
 pytestmark = pytest.mark.fr
 
 
+def test_streaming_latency_ms_clamps_negative_duration():
+    assert _streaming_latency_ms(10.0, 9.0) == 0
+
+
 @pytest.fixture(autouse=True)
 def _cleanup_safety_handlers():
     """Ensure safety handlers are cleared after each test in this file.
@@ -419,7 +424,7 @@ def test_anthropic_sync_stream_sets_streaming_latency_span_attrs(monkeypatch):
 
     timestamps = iter([101.25, 103.0])
     monkeypatch.setattr(
-        "opentelemetry.instrumentation.anthropic.streaming.time.time",
+        "opentelemetry.instrumentation.anthropic.streaming.time.perf_counter",
         lambda: next(timestamps),
     )
 
@@ -464,7 +469,7 @@ def test_anthropic_sync_stream_without_token_omits_streaming_latency_attrs(monke
 
     timestamps = iter([301.0])
     monkeypatch.setattr(
-        "opentelemetry.instrumentation.anthropic.streaming.time.time",
+        "opentelemetry.instrumentation.anthropic.streaming.time.perf_counter",
         lambda: next(timestamps),
     )
 
@@ -540,7 +545,7 @@ async def test_anthropic_async_stream_sets_streaming_latency_span_attrs(monkeypa
 
     timestamps = iter([201.1, 202.4])
     monkeypatch.setattr(
-        "opentelemetry.instrumentation.anthropic.streaming.time.time",
+        "opentelemetry.instrumentation.anthropic.streaming.time.perf_counter",
         lambda: next(timestamps),
     )
 

packages/opentelemetry-instrumentation-bedrock/opentelemetry/instrumentation/bedrock/__init__.py

@@ -263,7 +263,7 @@ def with_instrumentation(*args, **kwargs):
         # the response stream is fully consumed (which is why
         # ``start_as_current_span`` cannot be used directly).
         kwargs = _apply_invoke_prompt_safety(span, kwargs, _BEDROCK_INVOKE_SPAN_NAME)
-        stream_start_time = time.time()
+        stream_start_time = time.perf_counter()
         with trace.use_span(span, end_on_exit=False):
             response = fn(*args, **kwargs)
         _handle_stream_call(
@@ -306,7 +306,7 @@ def with_instrumentation(*args, **kwargs):
         kwargs = _apply_converse_prompt_safety(span, kwargs, _BEDROCK_CONVERSE_SPAN_NAME)
         # ST-10.4: see _instrumented_model_invoke_with_response_stream
         # for the rationale on use_span(end_on_exit=False).
-        stream_start_time = time.time()
+        stream_start_time = time.perf_counter()
         with trace.use_span(span, end_on_exit=False):
             response = fn(*args, **kwargs)
         if span.is_recording():

packages/opentelemetry-instrumentation-bedrock/opentelemetry/instrumentation/bedrock/streaming_safety.py

@@ -25,6 +25,10 @@
 FR_STREAMING_TIME_TO_GENERATE_MS = "fortifyroot.llm.streaming.time_to_generate_ms"
 
 
+def _streaming_latency_ms(start_time, end_time):
+    return round(max(0, end_time - start_time) * 1000)
+
+
 class _BedrockChunkStreamingSafety:
     def __init__(self, *, span, span_name: str, request_type: str):
         self._streams = CompletionTextStreamGroup(
@@ -157,7 +161,7 @@ def __init__(self, response, *, span, stream_done_callback=None, stream_start_ti
         self._self_accumulating_body = {}
         self._self_pending_event = None
         self._self_stream_start_time = (
-            time.time() if stream_start_time is None else stream_start_time
+            time.perf_counter() if stream_start_time is None else stream_start_time
         )
         self._self_first_token_time = None
         self._self_streaming_safety = _BedrockChunkStreamingSafety(
@@ -233,20 +237,20 @@ def _accumulate_events(self, payload):
     def _observe_streaming_text(self, text):
         if not isinstance(text, str) or not text or self._self_first_token_time is not None:
             return
-        now = time.time()
+        now = time.perf_counter()
         self._self_first_token_time = now
         if self._self_span.is_recording():
             self._self_span.set_attribute(
                 FR_STREAMING_TIME_TO_FIRST_TOKEN_MS,
-                round((now - self._self_stream_start_time) * 1000),
+                _streaming_latency_ms(self._self_stream_start_time, now),
             )
 
     def _finish_streaming_latency(self):
         if self._self_first_token_time is None or not self._self_span.is_recording():
             return
         self._self_span.set_attribute(
             FR_STREAMING_TIME_TO_GENERATE_MS,
-            round((time.time() - self._self_first_token_time) * 1000),
+            _streaming_latency_ms(self._self_first_token_time, time.perf_counter()),
         )
 
 
@@ -367,7 +371,7 @@ def __init__(
         self._self_response_msg = []
         self._self_span_ended = False
         self._self_stream_start_time = (
-            time.time() if stream_start_time is None else stream_start_time
+            time.perf_counter() if stream_start_time is None else stream_start_time
         )
         self._self_first_token_time = None
         self._self_streaming_safety = _BedrockConverseStreamingSafety(
@@ -449,20 +453,20 @@ def _observe_event(self, event):
     def _observe_streaming_text(self, text):
         if not isinstance(text, str) or not text or self._self_first_token_time is not None:
             return
-        now = time.time()
+        now = time.perf_counter()
         self._self_first_token_time = now
         if self._self_span.is_recording():
             self._self_span.set_attribute(
                 FR_STREAMING_TIME_TO_FIRST_TOKEN_MS,
-                round((now - self._self_stream_start_time) * 1000),
+                _streaming_latency_ms(self._self_stream_start_time, now),
             )
 
     def _finish_streaming_latency(self):
         if self._self_first_token_time is None or not self._self_span.is_recording():
             return
         self._self_span.set_attribute(
             FR_STREAMING_TIME_TO_GENERATE_MS,
-            round((time.time() - self._self_first_token_time) * 1000),
+            _streaming_latency_ms(self._self_first_token_time, time.perf_counter()),
         )
 
 

packages/opentelemetry-instrumentation-bedrock/tests/test_safety_hooks.py

@@ -20,6 +20,7 @@
     _decode_chunk_event,
     _encode_chunk_event,
     _set_payload_text,
+    _streaming_latency_ms,
     create_converse_stream_wrapper,
     create_invoke_stream_wrapper,
 )
@@ -53,6 +54,10 @@
 pytestmark = pytest.mark.fr
 
 
+def test_streaming_latency_ms_clamps_negative_duration():
+    assert _streaming_latency_ms(10.0, 9.0) == 0
+
+
 def setup_function():
     clear_safety_handlers()
     clear_completion_safety_stream_factory()
@@ -411,7 +416,7 @@ def test_invoke_streaming_wrapper_sets_latency_span_attrs(monkeypatch):
     )
     ticks = iter([100.125, 101.250])
     monkeypatch.setattr(
-        "opentelemetry.instrumentation.bedrock.streaming_safety.time.time",
+        "opentelemetry.instrumentation.bedrock.streaming_safety.time.perf_counter",
         lambda: next(ticks),
     )
 
@@ -496,7 +501,7 @@ def test_converse_streaming_wrapper_sets_latency_span_attrs(monkeypatch):
     )
     ticks = iter([200.050, 200.900])
     monkeypatch.setattr(
-        "opentelemetry.instrumentation.bedrock.streaming_safety.time.time",
+        "opentelemetry.instrumentation.bedrock.streaming_safety.time.perf_counter",
         lambda: next(ticks),
     )
 

packages/opentelemetry-instrumentation-langchain/opentelemetry/instrumentation/langchain/callback_handler.py

@@ -623,14 +623,14 @@ def on_llm_new_token(
         if span_holder.streaming_first_token_time is not None:
             return
 
-        first_token_time = time.time()
+        first_token_time = time.perf_counter()
         span_holder.streaming_first_token_time = first_token_time
         span = span_holder.span
         if span.is_recording():
             _set_span_attribute(
                 span,
                 FR_STREAMING_TIME_TO_FIRST_TOKEN_MS,
-                round((first_token_time - span_holder.start_time) * 1000),
+                round(max(0, first_token_time - span_holder.start_time) * 1000),
             )
 
     @dont_throw
@@ -735,15 +735,15 @@ def on_llm_end(
             set_chat_response(span, response)
 
         # Record duration before ending span
-        end_time = time.time()
+        end_time = time.perf_counter()
         streaming_first_token_time = self.spans[run_id].streaming_first_token_time
         if streaming_first_token_time is not None and span.is_recording():
             _set_span_attribute(
                 span,
                 FR_STREAMING_TIME_TO_GENERATE_MS,
-                round((end_time - streaming_first_token_time) * 1000),
+                round(max(0, end_time - streaming_first_token_time) * 1000),
             )
-        duration = end_time - self.spans[run_id].start_time
+        duration = max(0, end_time - self.spans[run_id].start_time)
         vendor = span.attributes.get(GenAIAttributes.GEN_AI_SYSTEM, "Langchain")
         self.duration_histogram.record(
             duration,

packages/opentelemetry-instrumentation-langchain/opentelemetry/instrumentation/langchain/span_utils.py

@@ -38,7 +38,7 @@ class SpanHolder:
     workflow_name: str
     entity_name: str
     entity_path: str
-    start_time: float = field(default_factory=time.time)
+    start_time: float = field(default_factory=time.perf_counter)
     request_model: Optional[str] = None
     streaming_first_token_time: Optional[float] = None
     # ST-10 review-round-2 fix (2026-05-11): every context_api.attach()