Skip to content

chore(vendor): sync OpenAI streaming latency span attrs #46

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sayonfortify merged 1 commit into from
Jun 18, 2026
Merged

chore(vendor): sync OpenAI streaming latency span attrs #46

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 75 additions & 11 deletions src/fortifyroot/_vendor/opentelemetry/instrumentation/openai/shared/chat_wrappers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,13 @@

LLM_REQUEST_TYPE = LLMRequestTypeValues.CHAT

# FR: stable internal span attributes carrying streaming latency as integer
# milliseconds, so the backend can extract TTFT/STTG into RDS llm_usage_events.
# Additive to (not a replacement for) the Mimir streaming histograms. Contract:
# fr-backend/docs/development/STREAMING_LATENCY_TTFT_STTG_PLAN.md
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"

logger = logging.getLogger(__name__)


Expand Down Expand Up @@ -773,13 +780,22 @@ def _process_item(self, item):
self._span.add_event(
name=f"{SpanAttributes.LLM_CONTENT_COMPLETION_CHUNK}")

if self._first_token and self._streaming_time_to_first_token:
if self._first_token:
self._time_of_first_token = time.time()
self._streaming_time_to_first_token.record(
self._time_of_first_token - self._start_time,
attributes=self._shared_attributes(),
)
if self._streaming_time_to_first_token:
self._streaming_time_to_first_token.record(
self._time_of_first_token - self._start_time,
attributes=self._shared_attributes(),
)
self._first_token = False
# FR: also persist TTFT (ms, int) on the span so the backend can
# extract it into RDS (the Mimir histogram above is unchanged).
if self._span and self._span.is_recording():
_set_span_attribute(
self._span,
FR_STREAMING_TIME_TO_FIRST_TOKEN_MS,
round((self._time_of_first_token - self._start_time) * 1000),
)

item = self._streaming_safety.process_chunk(item)
_accumulate_stream_items(item, self._complete_response)
Expand Down Expand Up @@ -847,6 +863,20 @@ def _process_complete_response(self):
_set_completions(
self._span, self._complete_response.get("choices"))

# FR: persist STTG (ms, int) on the span for RDS extraction, only
# when a first token actually arrived (mirrors the TTFT capture).
if (
self._span
and self._span.is_recording()
and not self._first_token
and self._time_of_first_token
):
_set_span_attribute(
self._span,
FR_STREAMING_TIME_TO_GENERATE_MS,
round((time.time() - self._time_of_first_token) * 1000),
)

self._span.set_status(Status(StatusCode.OK))
self._span.end()
self._cleanup_completed = True
Expand Down Expand Up @@ -950,11 +980,19 @@ def _build_from_streaming_response(

item_to_yield = streaming_safety.process_chunk(item)

if first_token and streaming_time_to_first_token:
if first_token:
time_of_first_token = time.time()
streaming_time_to_first_token.record(
time_of_first_token - start_time)
if streaming_time_to_first_token:
streaming_time_to_first_token.record(
time_of_first_token - start_time)
first_token = False
# FR: also persist TTFT (ms, int) on the span for RDS extraction.
if span and span.is_recording():
_set_span_attribute(
span,
FR_STREAMING_TIME_TO_FIRST_TOKEN_MS,
round((time_of_first_token - start_time) * 1000),
)

_accumulate_stream_items(item, complete_response)

Expand Down Expand Up @@ -990,6 +1028,15 @@ def _build_from_streaming_response(
if streaming_time_to_generate and time_of_first_token:
streaming_time_to_generate.record(time.time() - time_of_first_token)

# FR: persist STTG (ms, int) on the span for RDS extraction, only when a
# first token actually arrived (not first_token).
if span and span.is_recording() and not first_token and time_of_first_token:
_set_span_attribute(
span,
FR_STREAMING_TIME_TO_GENERATE_MS,
round((time.time() - time_of_first_token) * 1000),
)

_set_response_attributes(span, complete_response)
if should_emit_events():
for choice in complete_response.get("choices", []):
Expand Down Expand Up @@ -1027,11 +1074,19 @@ async def _abuild_from_streaming_response(

item_to_yield = streaming_safety.process_chunk(item)

if first_token and streaming_time_to_first_token:
if first_token:
time_of_first_token = time.time()
streaming_time_to_first_token.record(
time_of_first_token - start_time)
if streaming_time_to_first_token:
streaming_time_to_first_token.record(
time_of_first_token - start_time)
first_token = False
# FR: also persist TTFT (ms, int) on the span for RDS extraction.
if span and span.is_recording():
_set_span_attribute(
span,
FR_STREAMING_TIME_TO_FIRST_TOKEN_MS,
round((time_of_first_token - start_time) * 1000),
)

_accumulate_stream_items(item, complete_response)

Expand Down Expand Up @@ -1067,6 +1122,15 @@ async def _abuild_from_streaming_response(
if streaming_time_to_generate and time_of_first_token:
streaming_time_to_generate.record(time.time() - time_of_first_token)

# FR: persist STTG (ms, int) on the span for RDS extraction, only when a
# first token actually arrived (not first_token).
if span and span.is_recording() and not first_token and time_of_first_token:
_set_span_attribute(
span,
FR_STREAMING_TIME_TO_GENERATE_MS,
round((time.time() - time_of_first_token) * 1000),
)

_set_response_attributes(span, complete_response)
if should_emit_events():
for choice in complete_response.get("choices", []):
Expand Down
Loading