diff --git a/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/responses_wrappers.py b/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/responses_wrappers.py index 88aaca8034..5507dddb90 100644 --- a/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/responses_wrappers.py +++ b/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/responses_wrappers.py @@ -3,6 +3,7 @@ import re import threading import time +from collections import OrderedDict from typing import Any, Optional, Union from openai import AsyncStream, Stream @@ -198,6 +199,34 @@ class Config: responses: dict[str, TracedData] = {} +# Bounded set of response ids that already emitted a completion span. +# Holding only the id keeps memory tiny; evicting an id only reintroduces +# today's duplicate-span behavior (#4473). +_emitted_response_ids: OrderedDict[str, None] = OrderedDict() +_emitted_response_ids_lock = threading.Lock() +_EMITTED_RESPONSE_IDS_MAX = 2048 + + +def _mark_response_emitted(response_id: Optional[str]) -> bool: + """Return True on first completion emission for response_id. + + Later create/retrieve/parse calls on the same completed response skip + span emission so start_time/input/tools/trace_context from the original + create are not replaced by a degraded duplicate span. Lookup, LRU + update, insert, and eviction run under one lock so concurrent + completed wrappers cannot both observe "first emission". + """ + if not response_id: + return False + with _emitted_response_ids_lock: + if response_id in _emitted_response_ids: + _emitted_response_ids.move_to_end(response_id) + return False + _emitted_response_ids[response_id] = None + while len(_emitted_response_ids) > _EMITTED_RESPONSE_IDS_MAX: + _emitted_response_ids.popitem(last=False) + return True + def _derive_finish_reason(traced_data: TracedData) -> str: """Derive finish_reason from response.status instead of fabricating from block types. @@ -634,17 +663,21 @@ def responses_get_or_create_wrapper(tracer: Tracer, wrapped, instance, args, kwa return response if parsed_response.status == "completed": - # Restore the original trace context to maintain trace continuity - ctx = traced_data.trace_context if traced_data.trace_context else context_api.get_current() - span = tracer.start_span( - SPAN_NAME, - kind=SpanKind.CLIENT, - start_time=int(traced_data.start_time), - context=ctx, - ) - _set_request_attributes(span, prepare_kwargs_for_shared_attributes(non_sentinel_kwargs), instance) - set_data_attributes(traced_data, span) - span.end() + if _mark_response_emitted(parsed_response.id): + # Restore the original trace context to maintain trace continuity + ctx = traced_data.trace_context if traced_data.trace_context else context_api.get_current() + span = tracer.start_span( + SPAN_NAME, + kind=SpanKind.CLIENT, + start_time=int(traced_data.start_time), + context=ctx, + ) + _set_request_attributes(span, prepare_kwargs_for_shared_attributes(non_sentinel_kwargs), instance) + set_data_attributes(traced_data, span) + span.end() + # Completed responses are terminal; drop the module-global entry so + # successful runs do not retain TracedData for the process lifetime. + responses.pop(parsed_response.id, None) return response @@ -802,17 +835,21 @@ async def async_responses_get_or_create_wrapper( return response if parsed_response.status == "completed": - # Restore the original trace context to maintain trace continuity - ctx = traced_data.trace_context if traced_data.trace_context else context_api.get_current() - span = tracer.start_span( - SPAN_NAME, - kind=SpanKind.CLIENT, - start_time=int(traced_data.start_time), - context=ctx, - ) - _set_request_attributes(span, prepare_kwargs_for_shared_attributes(non_sentinel_kwargs), instance) - set_data_attributes(traced_data, span) - span.end() + if _mark_response_emitted(parsed_response.id): + # Restore the original trace context to maintain trace continuity + ctx = traced_data.trace_context if traced_data.trace_context else context_api.get_current() + span = tracer.start_span( + SPAN_NAME, + kind=SpanKind.CLIENT, + start_time=int(traced_data.start_time), + context=ctx, + ) + _set_request_attributes(span, prepare_kwargs_for_shared_attributes(non_sentinel_kwargs), instance) + set_data_attributes(traced_data, span) + span.end() + # Completed responses are terminal; drop the module-global entry so + # successful runs do not retain TracedData for the process lifetime. + responses.pop(parsed_response.id, None) return response @@ -1066,9 +1103,26 @@ def _process_complete_response(self): responses[parsed_response.id] = self._traced_data - set_data_attributes(self._traced_data, self._span) - self._span.set_status(StatusCode.OK) - self._span.end() + # Only completed responses are terminal. In-progress entries must + # stay available for Responses.retrieve polling (CodeRabbit #4482). + # Check first-emission BEFORE ending the long-lived stream span so a + # retrieve/parse that already marked this response_id cannot race + # into a second completion export. + status = getattr(self._traced_data, "response_status", None) + response_id = getattr(self._traced_data, "response_id", None) + is_terminal_completed = status == "completed" and bool(response_id) + should_emit = True + if is_terminal_completed: + should_emit = _mark_response_emitted(response_id) + + if should_emit: + set_data_attributes(self._traced_data, self._span) + self._span.set_status(StatusCode.OK) + # Always close the stream span; discard attributes on duplicates. + if self._span: + self._span.end() + if is_terminal_completed: + responses.pop(response_id, None) self._cleanup_completed = True except Exception as e: diff --git a/packages/opentelemetry-instrumentation-openai/tests/offline_review_4482.py b/packages/opentelemetry-instrumentation-openai/tests/offline_review_4482.py new file mode 100644 index 0000000000..a7e0b3005f --- /dev/null +++ b/packages/opentelemetry-instrumentation-openai/tests/offline_review_4482.py @@ -0,0 +1,92 @@ +"""Offline regression checks for openllmetry #4482 review follow-up. + +Runs without the full OpenAI/OTel stack by validating source contracts and +the emitted-id helper logic in isolation. +""" +from __future__ import annotations + +import sys +from collections import OrderedDict +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] # packages/opentelemetry-instrumentation-openai +SRC = ( + ROOT / "opentelemetry/instrumentation/openai/v1/responses_wrappers.py" +).read_text(encoding="utf-8") + + +def _mark_response_emitted(response_id, emitted: OrderedDict, max_size: int = 2048) -> bool: + if not response_id: + return False + if response_id in emitted: + emitted.move_to_end(response_id) + return False + emitted[response_id] = None + while len(emitted) > max_size: + emitted.popitem(last=False) + return True + + +def test_source_has_emitted_guard() -> None: + assert "_mark_response_emitted" in SRC + assert "_EMITTED_RESPONSE_IDS_MAX" in SRC + assert "_emitted_response_ids_lock" in SRC + assert "with _emitted_response_ids_lock:" in SRC + # sync + async completed paths still pop after optional emission + assert SRC.count("responses.pop(parsed_response.id, None)") >= 2 + # stream path pops via local response_id only when completed + assert "is_terminal_completed" in SRC + assert "responses.pop(response_id, None)" in SRC + # must not unconditionally pop on any stream exit + assert "if getattr(self._traced_data, \"response_id\", None):\n responses.pop" not in SRC + + +def test_mark_emitted_helper() -> None: + emitted: OrderedDict[str, None] = OrderedDict() + assert _mark_response_emitted("resp_1", emitted) is True + assert _mark_response_emitted("resp_1", emitted) is False + assert _mark_response_emitted("resp_2", emitted) is True + assert _mark_response_emitted("", emitted) is False + assert _mark_response_emitted(None, emitted) is False + # eviction keeps helper best-effort + small: OrderedDict[str, None] = OrderedDict() + for i in range(5): + assert _mark_response_emitted(f"id_{i}", small, max_size=3) is True + assert len(small) <= 3 + # evicted id can be marked again + assert _mark_response_emitted("id_0", small, max_size=3) is True + + +def test_stream_checks_emission_before_span_end() -> None: + assert "should_emit = _mark_response_emitted(response_id)" in SRC + # emission decision must appear before span.end() in the completed stream path + stream_fn = SRC.split("def _process_complete_response")[1] + mark_at = stream_fn.find("_mark_response_emitted") + end_at = stream_fn.find("self._span.end()") + assert mark_at != -1 and end_at != -1 and mark_at < end_at + # duplicates still close the span and still pop the completed entry + assert "if should_emit:" in stream_fn + assert "if is_terminal_completed:" in stream_fn + assert "responses.pop(response_id, None)" in stream_fn + + +def main() -> int: + tests = [ + test_source_has_emitted_guard, + test_mark_emitted_helper, + test_stream_checks_emission_before_span_end, + ] + failed = 0 + for fn in tests: + try: + fn() + print(f"PASS {fn.__name__}") + except AssertionError as exc: + failed += 1 + print(f"FAIL {fn.__name__}: {exc}") + print(f"{len(tests) - failed} passed, {failed} failed") + return 1 if failed else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/packages/opentelemetry-instrumentation-openai/tests/traces/test_responses.py b/packages/opentelemetry-instrumentation-openai/tests/traces/test_responses.py index b08fa3925b..ae119f8340 100644 --- a/packages/opentelemetry-instrumentation-openai/tests/traces/test_responses.py +++ b/packages/opentelemetry-instrumentation-openai/tests/traces/test_responses.py @@ -1,4 +1,5 @@ import json +from pathlib import Path import pytest from pydantic import BaseModel @@ -944,3 +945,160 @@ def test_parse_response_passes_through_plain_response(): result = parse_response(plain) assert result is plain + + +def test_completed_responses_are_removed_from_global_dict(): + """Regression for #4473: completed Responses API entries must not remain + in the module-global `responses` dict after a successful span is emitted.""" + import threading + from types import SimpleNamespace + from unittest.mock import MagicMock + + from opentelemetry.instrumentation.openai.v1 import responses_wrappers as rw + + rid = "resp_completed_cleanup_unit" + traced = rw.TracedData( + start_time=1.0, + response_id=rid, + input="hello", + response_status="completed", + output_text="world", + ) + rw.responses[rid] = traced + assert rid in rw.responses + + complete = MagicMock() + complete.id = rid + complete.model = "gpt-4.1" + complete.status = "completed" + complete.incomplete_details = None + complete.usage = None + complete.output = [] + + stream = SimpleNamespace( + _cleanup_lock=threading.Lock(), + _cleanup_completed=False, + _complete_response_data=complete, + _output_text="world", + _traced_data=traced, + _span=MagicMock(), + _tracer=MagicMock(), + _request_kwargs={}, + _start_time=1.0, + ) + rw.ResponseStream._process_complete_response(stream) + assert rid not in rw.responses, "completed streaming entry must be popped" + stream._span.end.assert_called() + + # Source contract: sync/async completed paths also pop after span.end(). + src = Path(rw.__file__).read_text(encoding="utf-8") + assert src.count("responses.pop(parsed_response.id, None)") >= 2 + assert "responses.pop(response_id, None)" in src + + +def test_incomplete_stream_exit_preserves_response_entry(): + """CodeRabbit #4482: stream cleanup must not drop in-progress entries that + Responses.retrieve still needs for polling.""" + import threading + from types import SimpleNamespace + from unittest.mock import MagicMock + + from opentelemetry.instrumentation.openai.v1 import responses_wrappers as rw + + rid = "resp_in_progress_stream_unit" + traced = rw.TracedData( + start_time=1.0, + response_id=rid, + input="hello", + response_status="in_progress", + ) + rw.responses[rid] = traced + + incomplete = MagicMock() + incomplete.id = rid + incomplete.model = "gpt-4.1" + incomplete.status = "in_progress" + incomplete.incomplete_details = None + incomplete.usage = None + incomplete.output = [] + + stream = SimpleNamespace( + _cleanup_lock=threading.Lock(), + _cleanup_completed=False, + _complete_response_data=incomplete, + _output_text="", + _traced_data=traced, + _span=MagicMock(), + _tracer=MagicMock(), + _request_kwargs={}, + _start_time=1.0, + ) + rw.ResponseStream._process_complete_response(stream) + assert rid in rw.responses, "in-progress streaming entry must be retained" + stream._span.end.assert_called() + rw.responses.pop(rid, None) + + +def test_duplicate_completed_emission_is_skipped(): + """#4473 second bug: a later retrieve/parse on an already completed response + must not emit a second degraded span.""" + import threading + from types import SimpleNamespace + from unittest.mock import MagicMock + + from opentelemetry.instrumentation.openai.v1 import responses_wrappers as rw + + rid = "resp_dup_emit_unit" + rw._emitted_response_ids.clear() + assert rw._mark_response_emitted(rid) is True + assert rw._mark_response_emitted(rid) is False + assert rw._mark_response_emitted(rid) is False + # empty id never marks + assert rw._mark_response_emitted("") is False + assert rw._mark_response_emitted(None) is False + + # Stream path: already-emitted completed response still closes the span + # and pops the dict entry, but does not re-export completion attributes. + traced = rw.TracedData( + start_time=1.0, + response_id=rid, + input="hello", + response_status="completed", + output_text="world", + ) + rw.responses[rid] = traced + complete = MagicMock() + complete.id = rid + complete.model = "gpt-4.1" + complete.status = "completed" + complete.incomplete_details = None + complete.usage = None + complete.output = [] + span = MagicMock() + stream = SimpleNamespace( + _cleanup_lock=threading.Lock(), + _cleanup_completed=False, + _complete_response_data=complete, + _output_text="world", + _traced_data=traced, + _span=span, + _tracer=MagicMock(), + _request_kwargs={}, + _start_time=1.0, + ) + rw.ResponseStream._process_complete_response(stream) + span.end.assert_called() + assert rid not in rw.responses + rw._emitted_response_ids.clear() + + +def test_cancel_paths_still_pop_responses_dict(): + from opentelemetry.instrumentation.openai.v1 import responses_wrappers as rw + + rid = "resp_cancel_unit" + traced = rw.TracedData(start_time=1.0, response_id=rid, input="x") + rw.responses[rid] = traced + existing = rw.responses.pop(rid, None) + assert existing is traced + assert rid not in rw.responses +