Share the with_raw_response proxy between instrumentations - #392
Draft
lmolkova wants to merge 4 commits into
Draft
Share the with_raw_response proxy between instrumentations#392lmolkova wants to merge 4 commits into
lmolkova wants to merge 4 commits into
Conversation
Move the stream-manager lifecycle, the close-finalizing response proxy, and the stream rebind helper into opentelemetry.util.genai.stream, and onboard the anthropic and openai instrumentations onto them. Also finalize telemetry for streams that expose aclose instead of close. Assisted-by: Claude Opus 5
The read hooks and parse() each set their own in-flight flag, and the two fallbacks that consult them checked different subsets, which read like a bug and was load-bearing. Replace both with one depth counter and one predicate, and cover the nesting that had no tests: a read inside a read, the read parse() drives, and a close arriving during either. Assisted-by: Claude Opus 5
Assisted-by: Claude Opus 5
Contributor
There was a problem hiding this comment.
Pull request overview
Centralizes raw-response and stream lifecycle handling in opentelemetry-util-genai, adopting it across OpenAI and Anthropic instrumentations.
Changes:
- Adds shared raw-response proxy state management.
- Shares stream manager and close-finalization helpers.
- Expands regression coverage for abandoned and asynchronously closed streams.
Reviewed changes
Copilot reviewed 28 out of 28 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
util/opentelemetry-util-genai/tests/test_stream.py |
Tests shared stream lifecycle helpers. |
util/opentelemetry-util-genai/tests/test_raw_response.py |
Tests raw-response proxy state transitions. |
util/opentelemetry-util-genai/src/opentelemetry/util/genai/stream.py |
Adds shared stream managers and close handling. |
util/opentelemetry-util-genai/src/opentelemetry/util/genai/raw_response.py |
Implements the shared raw-response proxy. |
util/opentelemetry-util-genai/.changelog/392.added |
Records the raw-response API addition. |
util/opentelemetry-util-genai/.changelog/390.changed |
Records aclose support. |
util/opentelemetry-util-genai/.changelog/390.added |
Records shared stream helpers. |
instrumentation/opentelemetry-instrumentation-google-genai/tests/generate_content/test_async_streaming.py |
Tests abandoned async stream finalization. |
instrumentation/opentelemetry-instrumentation-google-genai/.changelog/390.fixed |
Records the Google stream fix. |
instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_responses.py |
Updates raw HTTP streaming expectations. |
instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_response_wrappers.py |
Adapts stream-manager tests. |
instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_response_extractors.py |
Removes extractor-driven raw parsing tests. |
instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_raw_response_proxy.py |
Expands OpenAI raw-response coverage. |
instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_chat_completions.py |
Tests abandoned raw streams. |
instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_async_responses.py |
Updates async streaming expectations. |
instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/response_wrappers.py |
Adopts shared stream helpers. |
instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/response_extractors.py |
Stops parsing raw responses during extraction. |
instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/patch.py |
Routes chat raw responses through the proxy. |
instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/patch_responses.py |
Routes Responses API raw results. |
instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py |
Implements OpenAI-specific proxy hooks. |
instrumentation/opentelemetry-instrumentation-genai-openai/.changelog/392.fixed |
Records OpenAI raw-response fixes. |
instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_sync_messages.py |
Adapts synchronous raw-response tests. |
instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_wrappers.py |
Adapts asynchronous wrapper tests. |
instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_messages.py |
Adapts async raw-response tests. |
instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/wrappers.py |
Adopts shared stream helpers. |
instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/patch.py |
Routes Anthropic raw responses. |
instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/_raw_response.py |
Replaces duplicated proxy state logic. |
instrumentation/opentelemetry-instrumentation-genai-anthropic/.changelog/392.fixed |
Records Anthropic stream telemetry fix. |
Suppressed comments (2)
util/opentelemetry-util-genai/src/opentelemetry/util/genai/raw_response.py:219
- The asynchronous read path has the same leak as the synchronous path: when
aread()raises after a nested close was suppressed, execution never reaches_finalize_read_fallback(), so no later event can end the invocation. Finalize it as a failure and re-raise the original exception.
content = await original(*args, **kwargs)
finally:
self._self_reads_in_flight -= 1
self._finalize_read_fallback()
util/opentelemetry-util-genai/src/opentelemetry/util/genai/raw_response.py:349
RawResponseProxy.parse()is public, so acceptingAnyhere disables argument checking for consumers and violates the util package's public API rule. The forwarded values do not needAny;objectpreserves the open-ended SDK signature without suppressing type checking.
def parse(self, *args: Any, **kwargs: Any) -> object:
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Comment on lines
+201
to
+204
| content = original(*args, **kwargs) | ||
| finally: | ||
| self._self_reads_in_flight -= 1 | ||
| self._finalize_read_fallback() |
Comment on lines
+312
to
+315
| elif is_async_stream_wrapper(wrapper): | ||
| try: | ||
| await wrapper.close() | ||
| except Exception: # pylint: disable=broad-exception-caught |
Comment on lines
+62
to
+64
| http_response: Any | ||
|
|
||
| def parse(self, *args: Any, **kwargs: Any) -> Any: ... |
Comment on lines
+333
to
+335
| wrapped = object.__getattribute__(self, "__wrapped__") | ||
| if name in ("close", "aclose") and hasattr(wrapped, name): | ||
| return self._close |
Pull request dashboard statusWaiting on the author · refreshed 2026-08-20 15:38 UTC Move out of draft to request review. Status above doesn't look right?
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
The
with_raw_responseproxy landed for anthropic in #381 and openai had its own simpler copy, which had since diverged. This moves the state machine intoopentelemetry.util.genai.raw_responseand onboards both packages onto it.Fixes three openai bugs as a consequence: a
with_raw_responsestream the caller abandons emitted no span at all;parse(to=...)returned a memoized value of the wrong type; and instrumentation calledresult.parse()on the caller's behalf, applying their cast target and populating the SDK's parse cache.Routing between "completed payload" and "stream to be parsed" now uses the request's own
stream=argument rather than sniffing response headers, and the two read-in-flight flags collapse into a single depth counter.Stacked on #390 — the first commit belongs to that PR and drops out of this diff once it merges.