Skip to content

fix(openai): finalize completion spans on early stream close #4394

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

Open
CTWalk wants to merge 1 commit into
base: main
Choose a base branch
from
+101 −24
Open

fix(openai): finalize completion spans on early stream close #4394

Show file tree
Hide file tree
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
54 changes: 30 additions & 24 deletions ...instrumentation-openai/opentelemetry/instrumentation/openai/shared/completion_wrappers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -208,43 +208,49 @@ def _set_output_messages(span, choices):
@dont_throw
def _build_from_streaming_response(span, request_kwargs, response):
complete_response = {"choices": [], "model": "", "id": ""}
for item in response:
yield item
_accumulate_streaming_response(complete_response, item)
try:
for item in response:
yield item
_accumulate_streaming_response(complete_response, item)

_set_response_attributes(span, complete_response)
_set_response_attributes(span, complete_response)

_set_token_usage(span, request_kwargs, complete_response)
_set_token_usage(span, request_kwargs, complete_response)

if should_emit_events():
_emit_streaming_response_events(complete_response)
else:
if should_send_prompts():
_set_completions(span, complete_response.get("choices"))
if should_emit_events():
_emit_streaming_response_events(complete_response)
else:
if should_send_prompts():
_set_completions(span, complete_response.get("choices"))

span.set_status(Status(StatusCode.OK))
span.end()
span.set_status(Status(StatusCode.OK))
finally:
if span.is_recording():
span.end()

Comment on lines 209 to 230

@coderabbitai coderabbitai Bot Aug 1, 2026 •
edited
Loading

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== files =="
git ls-files | rg 'opentelemetry-instrumentation-openai|completion_wrappers|pyproject|requirements|lock' || true

echo "== target section =="
sed -n '180,270p' packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/completion_wrappers.py

echo "== imports and response type mentions =="
sed -n '1,80p' packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/completion_wrappers.py
rg -n "Stream|AsyncStream|close\(|aclose\(|_build_from_streaming_response|_abuild_from_streaming_response" packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai -S

Repository: traceloop/openllmetry

Length of output: 50377

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== package dependency declarations =="
sed -n '1,220p' packages/opentelemetry-instrumentation-openai/pyproject.toml
if [ -f packages/opentelemetry-instrumentation-openai/uv.lock ]; then
  rg -n 'name = "openai"|name = "httpx"|openai|httpx' packages/opentelemetry-instrumentation-openai/uv.lock | head -80
fi

echo "== shared is_streaming_response implementation =="
sed -n '360,395p' packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/__init__.py

echo "== async openai generator close semantics with Python's inspect =="
python3 - <<'PY'
import asyncio
from inspect import isasyncgenfunction, aclose

closed = False

async def stream():
    try:
        while True:
            yield 1
    finally:
        global closed
        closed = True

gen = stream()
async def run_close_abclose():
    try:
        await gen.__anext__()
    except StopAsyncIteration:
        pass
    await gen.aclose()
    print("aclose closed:", closed)

asyncio.run(run_close_abclose())
print("isasyncgenfunction:", isasyncgenfunction(stream))
print("has_aclose:", hasattr(gen, 'aclose'))
PY

Repository: traceloop/openllmetry

Length of output: 4653

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== openai 1.99.7 package location / installed files =="
uv venv --quiet .tmp-verify-venv 2>/dev/null || true
uv pip install --quiet -p .tmp-verify-venv openai==1.99.7 -c packages/opentelemetry-instrumentation-openai/uv.lock --no-deps 2>&1 | head -50 || true

python3 - <<'PY'
import os, tarfile, tempfile, urllib.request, pathlib
base = pathlib.Path(os.getcwd())
p = base / "packages/opentelemetry-instrumentation-openai/uv.lock"
data = p.read_text()
start = data.find('\nname = "openai"\n')
end = data.find('\nname = "opentelemetry-instrumentation-openai"\n', start)
block = data[start:end]
print("openai_lock=")
print(block)
print("has_url=", 'url = "' in block)
url_line = next((l for l in block.splitlines() if 'url = "' in l), None)
url = url_line.split('url = "')[1].rstrip('"') if url_line else None
print("url=", url)
if url:
    with tempfile.TemporaryDirectory() as td:
        path = pathlib.Path(td) / "openai.whl"
        urllib.request.urlretrieve(url, path)
        with tarfile.open(path, "r:zip") as tf:
            names = [n for n in tf.getnames() if n.endswith(("Stream.py", "Stream.pyi"))]
            print("stream_files=", names)
            for n in names:
                print("==", n, "==")
                print(tf.extractfile(n).read().decode("utf-8", "replace")[:5000])
PY

echo "== local test references =="
rg -n "memory_leak|not_consumed|partial_consumption|GeneratorExit|Generator\.close|close\(\)|aclose\(\)" packages/opentelemetry-instrumentation-openai -S --iglob '*.py' | head -120

Repository: traceloop/openllmetry

Length of output: 5290

🌐 Web query:

openai python 1.99.7 Stream close AsyncStream aclose iterator close source

💡 Result:

In the OpenAI Python library, specifically regarding the AsyncStream class and its closing behavior, the following technical details apply: 1. Handling of aclose: In version 1.99.7 and earlier, AsyncStream did not expose an aclose method, which often led to an AttributeError when instrumentation libraries or standard async patterns attempted to use await stream.aclose [1][2]. This was subsequently addressed in later versions of the library by adding aclose as an alias for the existing close method to ensure compatibility with PEP 525 conventions and standard async cleanup patterns [1][2]. 2. Stream Closing Mechanism: The internal logic for closing an asynchronous stream is designed to ensure the underlying network connection is managed correctly [3][4]. The AsyncStream.stream generator includes a finally block that calls await response.aclose to guarantee the response is closed even if the consumer does not iterate to completion [3]. 3. Resource Management and Connection Draining: A known issue related to streaming involves connection management after the [DONE] event is received [5][4]. When an SSE stream reaches the [DONE] sentinel, the generator typically breaks out of its iteration loop [3][4]. If the remaining bytes—specifically the HTTP/1.1 chunked terminator—are not drained from the iterator before the connection is closed, it can result in a force-close of the connection rather than a graceful return to the connection pool [5][4]. Subsequent patches introduced logic to "drain" these remaining bytes in the finally block (using async for _ in iterator: pass) to prevent these connection-related issues [5]. Users experiencing issues with streaming lifecycle or connection errors are encouraged to ensure they are using a more recent version of the library where these compatibility and connection-draining fixes have been integrated [5].

Citations:

🌐 Web query:

openai-python issue 763 unclosed streams leak connections

💡 Result:

Issue #763 in the openai-python GitHub repository addressed a bug where network connections were not being returned to the HTTP connection pool when using streaming requests [1]. This resulted in connection exhaustion (often manifesting as APITimeoutError or httpx.PoolTimeout) in high-concurrency environments because the underlying connections remained occupied [1][2]. The issue was resolved in version 1.2.3 of the OpenAI Python SDK [1]. For users encountering similar behaviors or connection management issues, the recommended practices include: 1. Manual Response Closure: If you are streaming responses and not consuming the entire stream, you can explicitly close the response object using await stream.response.aclose() (for async) or stream.response.close() (for sync) to ensure the connection is returned to the pool [1][3][4]. 2. Use Context Managers: When working with the client, using it within a context manager (async with AsyncOpenAI() as client: or with OpenAI() as client:) ensures that the client and its underlying connection pool are cleaned up appropriately when the block exits [5][6]. 3. Client Reuse: The SDK is designed for the client instance to be reused across requests rather than instantiated repeatedly. Creating new clients per request can lead to resource leaks and socket exhaustion [5][6][7]. 4. Adjust Pool Limits: In high-scale production environments, you may need to explicitly configure the connection pool limits to be larger than the defaults [8][7]. This can be done by passing a custom http_client to the OpenAI client constructor: from openai import AsyncOpenAI import httpx client = AsyncOpenAI( http_client=httpx.AsyncClient( limits=httpx.Limits(max_connections=1000, max_keepalive_connections=100))) [7] These approaches address the connection leaks and pool timeouts reported in issue #763 and related discussions [1][2][7].

Citations:

🌐 Web query:

openai python 1.99.7 AsyncStream aclose attribute source code

💡 Result:

In the OpenAI Python library, specifically as of version 1.99.7, the AsyncStream class contains an aclose method [1][2][3]. Historically, AsyncStream only provided a close method, which led to an AttributeError when external instrumentation tools (such as Langfuse) or standard Python async cleanup patterns expected an aclose method—a convention established by PEP 525 for async generators and used by libraries like httpx and asyncio [1][2][4]. This issue was addressed by adding aclose as an alias for the existing close method, ensuring compatibility with standard asynchronous resource cleanup workflows [1][2]. In the source code (src/openai/_streaming.py), the implementation essentially delegates the cleanup to the underlying HTTP response object [5][6][7]: async def aclose(self) -> None: await self.close async def close(self) -> None: """ Close the response and release the connection. Automatically called if the response body is read to completion. """ await self.response.aclose The primary purpose of this change was to satisfy the PEP 525 async cleanup contract, preventing crashes in environments where wrappers or proxy objects might attempt to call aclose on the stream [1][2][4].

Citations:

🌐 Web query:

langfuse github close async streaming openai Python AttributeError 1.99.7 aclose

💡 Result:

The AttributeError: 'AsyncStream' object has no attribute 'aclose' that occurred in older versions of the Langfuse Python SDK (such as around version 1.99.7) was caused by an incompatibility between Langfuse’s streaming adapter and the OpenAI Python SDK's async stream management [1][2][3]. When using Langfuse to instrument OpenAI's async streaming API—particularly with structured outputs—the Langfuse wrapper was historically passed as the raw stream object to OpenAI's internal stream manager [1][4]. OpenAI’s stream manager would then attempt to call.aclose on the underlying response object [1][5][4]. Because the Langfuse adapter was incorrectly exposing an AsyncStream object (which lacked the aclose method) as the response, the Python interpreter raised an AttributeError [1][3][5]. This issue was addressed through several rounds of improvements: 1. Initial Hotfixes: Earlier versions of Langfuse and the OpenAI SDK worked to resolve the mismatch by ensuring the adapter properly delegated cleanup or by adding aclose as an alias for close on the OpenAI AsyncStream class itself to better comply with PEP 525 async generator conventions [5][6][7]. 2. Robust Architectural Change: In more recent updates (such as those represented in PR #1627, merged in April 2026), Langfuse moved away from wrapping the response in a custom generator class that replaced the native OpenAI stream objects [8][9]. Instead, the SDK now preserves the native v1 stream contract by monkey-patching the existing OpenAI Stream/AsyncStream objects to collect data while keeping the original methods (like close and aclose) intact [8][9]. If you are encountering this error, it is recommended to update the langfuse package to the latest version, as this architectural change ensures compatibility with the native OpenAI client contract [8][9].

Citations:

Close the OpenAI stream before ending the span. Both _build_from_streaming_response and _abuild_from_streaming_response only call span.end() in finally; if the wrapper generator exits before full consumption, OpenAI’s stream iterator is abandoned without returning its HTTP connection to the client pool. Add a raw stream cleanup path before ending the span: call response.close() for sync Stream responses and await response.aclose()/response.aclose() for async AsyncStream responses, with an appropriate guard if the wrapper is used with streams that may not expose the async method.

📍 Affects 1 file
  • packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/completion_wrappers.py#L209-L230 (this comment)
  • packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/completion_wrappers.py#L235-L253
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In
`@packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/completion_wrappers.py`
around lines 209 - 230, Update both _build_from_streaming_response and
_abuild_from_streaming_response to close the raw response in their cleanup paths
before ending the span: call response.close() for synchronous streams and
guarded response.aclose() for asynchronous streams, awaiting it in the async
wrapper when available. Preserve span.end() execution even if stream cleanup is
unavailable or fails.

@CTWalk CTWalk Aug 1, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks — I've left this out of scope rather than folding it in. The change
alters when the span is ended, not when the stream is released; an
early-closed Completions stream was abandoned the same way before it. The chat
path from #3155 has the same gap on its __del__ → _ensure_cleanup
abandonment path, so it looks repo-wide rather than specific to this file.

One correction for anyone acting on the suggestion: on openai 1.99.7,
AsyncStream has no aclose. The async close is await response.close(), and
Stream.close() is sync.

Happy to open a separate PR for connection cleanup across both paths if that
would be useful.

@coderabbitai coderabbitai Bot Aug 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@CTWalk, agreed. This pull request changes span finalization only. It does not introduce the raw stream cleanup gap.

Thank you for the openai 1.99.7 correction. A follow-up must use await response.close() for AsyncStream and response.close() for Stream.

Would you like me to open a GitHub issue for repository-wide stream connection cleanup across the Completions and chat wrapper paths?

🐇 💭

✏️ Learnings added 
Learnt from: CTWalk
Repo: traceloop/openllmetry PR: 4394
File: packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/completion_wrappers.py:209-230
Timestamp: 2026-08-01T04:11:27.508Z
Learning: For the OpenAI Python SDK version 1.99.7, `AsyncStream` does not provide `aclose()`. Async cleanup uses `await response.close()`, while synchronous `Stream` cleanup uses `response.close()`. Any future connection-cleanup work in the OpenAI instrumentation wrappers must use these methods.

Note: Learnings are effective only in the context of similar code segments. To apply general coding guidelines, please configure review instructions. You can manage existing learnings in the app.

You are interacting with an AI system.

@CTWalk CTWalk Aug 1, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not for now — worth measuring whether connections are actually held open before filing anything. If it holds up I'll raise it separately with a reproduction.

@coderabbitai coderabbitai Bot Aug 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@CTWalk, acknowledged. I will not open an issue now.

Please raise a separate issue if measurement confirms that early stream abandonment retains connections. Include a minimal reproduction if available.

🧠 Learnings used 
Learnt from: CTWalk
Repo: traceloop/openllmetry PR: 4394
File: packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/completion_wrappers.py:209-230
Timestamp: 2026-08-01T04:11:27.508Z
Learning: For OpenAI Python SDK version 1.99.7, do not call `aclose()` on `AsyncStream`, because it does not provide that method. In OpenAI instrumentation wrappers, use `await response.close()` for asynchronous cleanup and `response.close()` for synchronous `Stream` cleanup.

You are interacting with an AI system.


@dont_throw
async def _abuild_from_streaming_response(span, request_kwargs, response):
complete_response = {"choices": [], "model": "", "id": ""}
async for item in response:
yield item
_accumulate_streaming_response(complete_response, item)
try:
async for item in response:
yield item
_accumulate_streaming_response(complete_response, item)

_set_response_attributes(span, complete_response)
_set_response_attributes(span, complete_response)

_set_token_usage(span, request_kwargs, complete_response)
_set_token_usage(span, request_kwargs, complete_response)

if should_emit_events():
_emit_streaming_response_events(complete_response)
else:
if should_send_prompts():
_set_completions(span, complete_response.get("choices"))
if should_emit_events():
_emit_streaming_response_events(complete_response)
else:
if should_send_prompts():
_set_completions(span, complete_response.get("choices"))

span.set_status(Status(StatusCode.OK))
span.end()
span.set_status(Status(StatusCode.OK))
finally:
if span.is_recording():
span.end()


def _emit_streaming_response_events(complete_response):
Expand Down
71 changes: 71 additions & 0 deletions packages/opentelemetry-instrumentation-openai/tests/traces/test_completions.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import gc
from unittest.mock import patch

import httpx
Expand Down Expand Up @@ -393,6 +394,55 @@ def test_completion_streaming(
), "Assert that it doesn't emit logs when use_legacy_attributes is True"


@pytest.mark.vcr
@pytest.mark.default_cassette("test_completion_streaming.yaml")
def test_completion_streaming_early_close(
instrument_legacy, span_exporter, log_exporter, mock_openai_client
):
response = mock_openai_client.completions.create(
model="davinci-002",
prompt="Tell me a joke about opentelemetry",
stream=True,
)

next(response)
response.close()

spans = span_exporter.get_finished_spans()
assert [span.name for span in spans] == [
"openai.completion",
]


@pytest.mark.vcr
@pytest.mark.default_cassette("test_completion_streaming.yaml")
def test_completion_streaming_exception_during_consumption(
instrument_legacy, span_exporter, log_exporter, mock_openai_client
):
"""The span is still ended when the caller aborts consumption by raising."""

response = mock_openai_client.completions.create(
model="davinci-002",
prompt="Tell me a joke about opentelemetry",
stream=True,
)

with pytest.raises(ValueError, match="simulated interruption"):
for _ in response:
raise ValueError("simulated interruption")

del response
gc.collect()

spans = span_exporter.get_finished_spans()
assert [span.name for span in spans] == [
"openai.completion",
]
# Status is deliberately not asserted: what an incompletely observed
# completion should report is a separate question from ending the span.
assert spans[0].end_time is not None


@pytest.mark.vcr
def test_completion_streaming_with_events_with_content(
instrument_with_content, span_exporter, log_exporter, openai_client
Expand Down Expand Up @@ -548,6 +598,27 @@ async def test_async_completion_streaming(
), "Assert that it doesn't emit logs when use_legacy_attributes is True"


@pytest.mark.vcr
@pytest.mark.default_cassette("test_async_completion_streaming.yaml")
@pytest.mark.asyncio
async def test_async_completion_streaming_early_close(
instrument_legacy, span_exporter, log_exporter, async_openai_client
):
response = await async_openai_client.completions.create(
model="davinci-002",
prompt="Tell me a joke about opentelemetry",
stream=True,
)

await anext(response)
await response.aclose()

spans = span_exporter.get_finished_spans()
assert [span.name for span in spans] == [
"openai.completion",
]


@pytest.mark.vcr
@pytest.mark.asyncio
async def test_async_completion_streaming_with_events_with_content(
Expand Down