fix(openai-responses): remove completed entries from the global responses dict

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:

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())

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
+