From 2f1e1c2f0007630be116a30c4a6f6f6f10b93b3b Mon Sep 17 00:00:00 2001
From: sayonfortify <sayon@fortifyroot.com>
Date: Sat, 20 Jun 2026 23:28:10 +0530
Subject: [PATCH] fix(anthropic): record streaming latency before safety
 masking

---
 .../instrumentation/anthropic/streaming.py    |  4 +-
 .../tests/test_safety_unit.py                 | 50 +++++++++++++++++++
 2 files changed, 52 insertions(+), 2 deletions(-)

diff --git a/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/streaming.py b/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/streaming.py
index 672701148a..8c34895928 100644
--- a/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/streaming.py
+++ b/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/streaming.py
@@ -304,7 +304,6 @@ def __next__(self):  # FR: rewritten for pending-item safety buffer
                     self._instrumentation_completed = True
                 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.perf_counter()
                 _set_streaming_first_token_latency(
@@ -312,6 +311,7 @@ def __next__(self):  # FR: rewritten for pending-item safety buffer
                     self._start_time,
                     self._first_token_time,
                 )
+            item = self._streaming_safety.process_item(item)  # FR: streaming safety processing
             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
@@ -508,7 +508,6 @@ async def __anext__(self):  # FR: rewritten for pending-item safety buffer
                     self._instrumentation_completed = True
                 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.perf_counter()
                 _set_streaming_first_token_latency(
@@ -516,6 +515,7 @@ async def __anext__(self):  # FR: rewritten for pending-item safety buffer
                     self._start_time,
                     self._first_token_time,
                 )
+            item = self._streaming_safety.process_item(item)  # FR: streaming safety processing
             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
diff --git a/packages/opentelemetry-instrumentation-anthropic/tests/test_safety_unit.py b/packages/opentelemetry-instrumentation-anthropic/tests/test_safety_unit.py
index e07b5d67c7..9433876f9d 100644
--- a/packages/opentelemetry-instrumentation-anthropic/tests/test_safety_unit.py
+++ b/packages/opentelemetry-instrumentation-anthropic/tests/test_safety_unit.py
@@ -463,6 +463,56 @@ def test_anthropic_sync_stream_sets_streaming_latency_span_attrs(monkeypatch):
     assert finished[0].attributes[FR_STREAMING_TIME_TO_GENERATE_MS] == 1750
 
 
+def test_anthropic_sync_stream_sets_latency_attrs_before_safety_masks_token(monkeypatch):
+    clear_safety_handlers()
+    register_completion_safety_stream_factory(lambda _: _FakeStreamSession())
+    exporter, tracer = _test_tracer()
+
+    timestamps = iter([101.25, 103.0])
+    monkeypatch.setattr(
+        "opentelemetry.instrumentation.anthropic.streaming.time.perf_counter",
+        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="secret"),
+                ),
+                SimpleNamespace(type="content_block_stop", index=0),
+            ]
+        ),
+        SimpleNamespace(count_tokens=lambda text: len(text)),
+        100.0,
+        kwargs={},
+    )
+
+    first = next(stream)
+    second = next(stream)
+    third = 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 second.delta.text == "masked-tail"
+    assert third.type == "content_block_stop"
+    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()