fix(openai): finalize completion spans on early stream close - #4394
fix(openai): finalize completion spans on early stream close#4394CTWalk wants to merge 1 commit into
Conversation
The Completions streaming wrappers end their spans after the iteration loop. That works when the stream is exhausted, but close() or aclose() exits the generator at its suspended yield, before span.end() is reached, so an early-closed stream exports no openai.completion span at all. Wrap both streaming generator bodies in try/finally and end a still-recording span in finally, so exhaustion, close(), and aclose() converge on one terminal span.end(). Full-consumption response attributes, token usage, events, and StatusCode.OK still run only on normal exhaustion. Because finally also runs on exceptions, a stream that raises mid-iteration now ends and exports its span instead of leaving one recording and never exported. Add three tests covering sync early close, async early close, and an exception raised during consumption, the last mirroring the chat-side coverage added in traceloop#3155. All three reuse the existing streaming cassettes via pytest-recording's default_cassette marker, so no fixture data is added. Most of the diff is re-indentation from the new try block; the semantic change is six lines, visible with git diff -w. Related to traceloop#3151 and traceloop#3155, which established this contract for chat streams. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
|
|
📝 WalkthroughWalkthroughOpenAI streaming completion wrappers now end spans during normal completion, errors, and early interruption. Tests cover synchronous and asynchronous early close, plus synchronous consumption exceptions and garbage-collection cleanup. ChangesOpenAI streaming span finalization
Estimated code review effort: 3 (Moderate) | ~20 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (3)
packages/opentelemetry-instrumentation-openai/tests/traces/test_completions.py (2)
601-621: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winNo async counterpart for the consumption-exception test.
The PR adds a sync consumption-exception test (Lines 419-445) but no equivalent async version alongside this async early-close test. Since
_abuild_from_streaming_responsemirrors the sync implementation, an async exception-during-consumption test would close the coverage gap for that code path.🤖 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/tests/traces/test_completions.py` around lines 601 - 621, Add an async consumption-exception test alongside test_async_completion_streaming_early_close, using async_openai_client and the streaming completion response. Consume the response until an exception is raised, close it appropriately, and assert the finished span behavior matches the synchronous consumption-exception test, covering _abuild_from_streaming_response.
397-415: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winEarly-close test only checks span presence, not that success data is absent.
test_completion_streaming_early_closeconfirms the span is exported, but it does not assert that full-consumption-only data (response attributes, token usage,StatusCode.OK) is absent for an early-closed stream. Since that is the specific contract this PR establishes ("Early-closed streams retain collected request attributes without fabricated response data or success status"), a negative assertion here would directly pin down that contract for regression protection.🤖 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/tests/traces/test_completions.py` around lines 397 - 415, Extend test_completion_streaming_early_close to assert the exported span retains collected request attributes but has no response attributes or token-usage data and does not have StatusCode.OK after the stream is closed early. Keep the existing span-presence assertion and use the span’s established attribute and status accessors.packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/completion_wrappers.py (1)
211-253: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicate finalize logic between sync and async wrappers.
The post-loop sequence (
_set_response_attributes,_set_token_usage, event/completion emission,set_status(OK), and thefinallyguard) is duplicated verbatim between_build_from_streaming_responseand_abuild_from_streaming_response. Only the loop syntax (forvsasync for) differs.Extract the shared finalize step into a helper (e.g.,
_finalize_streaming_span(span, request_kwargs, complete_response)) called from both generators inside thetry. This also gives one place to add the stream-close fix suggested above.♻️ Sketch of the shared helper
+def _finalize_streaming_span(span, request_kwargs, complete_response): + _set_response_attributes(span, complete_response) + _set_token_usage(span, request_kwargs, complete_response) + if should_emit_events(): + _emit_streaming_response_events(complete_response) + elif should_send_prompts(): + _set_completions(span, complete_response.get("choices")) + span.set_status(Status(StatusCode.OK)) + + def _build_from_streaming_response(span, request_kwargs, response): complete_response = {"choices": [], "model": "", "id": ""} try: for item in response: yield item _accumulate_streaming_response(complete_response, item) - - _set_response_attributes(span, 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")) - - span.set_status(Status(StatusCode.OK)) + _finalize_streaming_span(span, request_kwargs, complete_response) finally: if span.is_recording(): span.end()🤖 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 211 - 253, Extract the duplicated post-stream processing from _build_from_streaming_response and _abuild_from_streaming_response into a shared _finalize_streaming_span helper, including response attributes, token usage, event or completion emission, and the OK status. Call the helper after each sync or async loop within its respective try block, while retaining the existing finally span.is_recording() guard and span.end() behavior in both generators.
🤖 Prompt for all review comments with 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.
Inline comments:
In
`@packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/completion_wrappers.py`:
- Around line 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.
---
Nitpick comments:
In
`@packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/completion_wrappers.py`:
- Around line 211-253: Extract the duplicated post-stream processing from
_build_from_streaming_response and _abuild_from_streaming_response into a shared
_finalize_streaming_span helper, including response attributes, token usage,
event or completion emission, and the OK status. Call the helper after each sync
or async loop within its respective try block, while retaining the existing
finally span.is_recording() guard and span.end() behavior in both generators.
In
`@packages/opentelemetry-instrumentation-openai/tests/traces/test_completions.py`:
- Around line 601-621: Add an async consumption-exception test alongside
test_async_completion_streaming_early_close, using async_openai_client and the
streaming completion response. Consume the response until an exception is
raised, close it appropriately, and assert the finished span behavior matches
the synchronous consumption-exception test, covering
_abuild_from_streaming_response.
- Around line 397-415: Extend test_completion_streaming_early_close to assert
the exported span retains collected request attributes but has no response
attributes or token-usage data and does not have StatusCode.OK after the stream
is closed early. Keep the existing span-presence assertion and use the span’s
established attribute and status accessors.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 04afb729-1298-4a21-b0c0-dd00f3b471e5
📒 Files selected for processing (2)
packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/completion_wrappers.pypackages/opentelemetry-instrumentation-openai/tests/traces/test_completions.py
| 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() | ||
|
|
There was a problem hiding this comment.
🩺 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 -SRepository: 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'))
PYRepository: 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 -120Repository: 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:
- 1: fix: add aclose() method to AsyncStream for PEP 525 compliance openai/openai-python#2880
- 2: AsyncStream missing aclose() causes AttributeError in streaming structured output openai/openai-python#2853
- 3: https://github.com/openai/openai-python/blob/main/src/openai/_streaming.py
- 4: **Streaming: connection force-closed (TCP FIN) after
[DONE]SSE event because chunked terminator is not drained — regression from6132922c** openai/openai-python#3440 - 5: fix: drain remaining stream bytes after [DONE] event to prevent connection force-close openai/openai-python#3510
🌐 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:
- 1: The connection is not returned to the httpx pool when using a stream openai/openai-python#763
- 2: Constant timeouts after multiple calls with async openai/openai-python#769
- 3: Connection Not Closing Properly When Using GPT API with Stream openai/openai-python#667
- 4: memory leak, consumes entire system after just days of usage openai/openai-python#1181
- 5: Unclosed client session openai/openai-python#570
- 6: Why docs do not use context manager for several examples openai/openai-python#2779
- 7: Unclear how to use the new AsyncClient at scale - memory leaks or timeouts openai/openai-python#874
- 8: Open AI Clients starts timing out after some time in production openai/openai-python#2688
🌐 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:
- 1: fix: add aclose() method to AsyncStream for PEP 525 compliance openai/openai-python#2880
- 2: AsyncStream missing aclose() causes AttributeError in streaming structured output openai/openai-python#2853
- 3: https://pypi.org/project/openai/1.99.7/
- 4: bug: OpenAI streaming structured output causes AttributeError in Python/FastAPI (revisited) langfuse/langfuse#5685
- 5: https://github.com/openai/openai-python/blob/5e8f09c2/src/openai/_streaming.py
- 6: https://github.com/openai/openai-python/blob/main/src/openai/_streaming.py
- 7: https://github.com/openai/openai-python/blob/722d3fff/src/openai/_streaming.py
🌐 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:
- 1: fix: ensure raw_stream.response is httpx.Response in langfuse streaming adapter langfuse/langfuse-python#1409
- 2: bug(sdk-js): LangfuseResponseGeneratorAsync incompatible with AsyncOpenAI.beta.chat.completions.stream() langfuse/langfuse#6368
- 3: bug: OpenAI streaming structured output causes AttributeError in Python/FastAPI langfuse/langfuse#5361
- 4: bug: OpenAI streaming structured output causes AttributeError in Python/FastAPI (revisited) langfuse/langfuse#5685
- 5: AsyncStream missing aclose() causes AttributeError in streaming structured output openai/openai-python#2853
- 6: fix: add aclose() method to AsyncStream for PEP 525 compliance openai/openai-python#2880
- 7: https://data.safetycli.com/packages/pypi/langfuse/changelog?page=76
- 8: langfuse/langfuse-python@7aa632f
- 9: fix(openai): preserve native v1 stream contract langfuse/langfuse-python#1627
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.
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
@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.
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
@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.
2 participants
Summary
Finalize
openai.completionspans when a caller closes a sync or asyncCompletions stream before consuming it fully.
This brings the Completions path in line with OpenLLMetry's existing cleanup
behavior for chat and Responses streams in ending the span; it deliberately
does not copy chat's
StatusCode.OK, as noted below. It also adds focused syncand async regression tests.
Related: #3151 and #3155.
Problem
The Completions streaming wrappers end their spans after the iteration loop.
That works when the stream is exhausted, but
close()oraclose()exits thegenerator at its suspended
yield, beforespan.end()is reached.Reproduced with
OpenAIandAsyncOpenAIclients against a local SSE serverand an in-memory span exporter on
main@93429cfb:openai.completionspansclose()aclose()The full-consumption controls show that the instrumentation and exporter are
working; the missing span is specific to early close.
A cancelled or timed-out completion is therefore invisible rather than
truncated: the request that was worth investigating is the one with no span at
all, and #3151 reports the same loss on the chat path as a bug.
Change
try/finallyand enda still-recording span in
finally, so exhaustion,close(), andaclose()converge on one terminal
span.end().events, and
StatusCode.OK— they still run only on normal exhaustion.during consumption — the last mirroring
test_chat_streaming_exception_during_consumptionfrom fix(openai): support for openai non-consumed streams #3155. All three reusethe existing streaming cassettes via
@pytest.mark.default_cassette, so nofixture data is added.
After the change, all four cases above export exactly one span.
Most of the diff in
completion_wrappers.pyis re-indentation from the newtryblock. The semantic change is six lines —git diff -wshows just thetry:,finally:, andif span.is_recording():on each of the two generators.An early-closed stream exports the request attributes already collected. It
does not fabricate response attributes, token usage, or a successful status for
a response that was not fully observed. This is the one place the change does
not follow chat:
ChatStreamreportsStatusCode.OKfor an abandoned stream,and asserting success for a response nobody finished reading seemed worth not
copying. Glad to align it either way if you'd rather the two paths match
exactly.
Because
finallyalso runs on exceptions, the same change covers a stream thatraises mid-iteration: the span is now ended and exported instead of being left
recording and never exported. Measured on
main@93429cfbwith a stream thatraises after one chunk — 0 spans exported before, 1 after, with the exception
propagating unchanged in both. The new consumption-exception test pins the
ending; it deliberately does not assert a status, since which status such a span
should carry is a semantic-convention question rather than a lifecycle one.
Verification
From
packages/opentelemetry-instrumentation-openai:All three new tests fail on the unpatched tree because no finished
openai.completionspan is exported.The 26 includes the two pre-existing invalid-key tests, which carry no cassette
and need a reachable endpoint to see the 401 they assert. They are unrelated to
this change; without network access they fail on both the patched and unpatched
tree alike.
Checklist
screenshots from an observability platform.
Not included: this lifecycle behavior is asserted directly with the
repository's in-memory exporter and does not change an observability UI.
Not applicable: there is no public API, configuration, or documented
output change.
AI assistance was used during investigation and draft preparation. The
resulting patch was reviewed against the source, reproduced on both paths, and
verified with the checks above.
Summary by CodeRabbit