fix(server-utils): End Anthropic stream spans drained through the raw Response - #24485
Draft
RulaKhaled wants to merge 5 commits into
Draft
RulaKhaled wants to merge 5 commits into
RulaKhaled wants to merge 5 commits into
Conversation
… Response Hook `Stream.fromSSEResponse` and wrap the `Response` body it is built from, so the `gen_ai.chat` span ends however the caller drains the stream. Every consumption path bottoms out in `response.body`, but only the SDK `Stream`'s async iterator was instrumented. A caller who takes `.asResponse()` or `.withResponse()` and reads the body themselves never touches that iterator, so nothing ever ended the span and it was dropped. `tee()` was lost the same way, since it calls `this.iterator()` directly. Fixes #24258 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Contributor
size-limit report 📦
|
…pping it The wrapper acquired a reader as soon as it was installed, and its stream defaulted to a high-water mark of 1, so it read a chunk ahead before anyone asked for one. Both disturb the body, which made `text()`, `arrayBuffer()` and `clone()` throw on a `.asResponse()` result the caller had not read yet. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
The body wrapper sees every chunk whichever way the caller drains the stream, so keeping the async-iterator instrumentation alongside it bought nothing and cost a claim handshake to stop the two from double-counting. Drop it from the channel path: one accumulator, one owner, and `wrapStreamResult` becomes a membership test. The manual `instrumentAnthropicAiClient` path still uses the iterator. A streaming call whose body we couldn't wrap now ends at `asyncEnd` with request attributes rather than hanging, so the degraded case stays graceful. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Handing the span to the SSE body wrapper left several paths uncovered.
A body that is not a web ReadableStream, which is what an injected
node-fetch or undici shim hands back, could not be wrapped at all, so the
span ended the moment create() resolved with request attributes only.
Patching the SDK Stream's iterator now serves as the fallback for those.
clone() tees the response's internal body and swaps in one branch, which
left the stream the wrapper had captured locked, so the next read threw.
The wrapper now resolves the source through the prototype getter on every
read instead of holding on to the stream it was handed.
text(), json() and arrayBuffer() read the internal body and never touch
the property we shadow, so the span never ended. Those are wrapped too,
and text() and arrayBuffer() feed their result through the accumulator so
the response attributes survive.
Three smaller ones: recordOutputs is resolved per call rather than once at
subscribe time, since subscribeToSseStream runs once per process and a
later client can carry different options. The frame parser's try now
covers a single frame, so one unparsable line no longer costs us the rest
of its chunk, where message_delta and message_stop ride. An unsampled span
skips the wrap entirely, and settle() flushes a trailing frame left
without a newline.
The pass-through is a byte stream now, so getReader({ mode: 'byob' })
keeps working on a response that supported it before.
instrumentRawSseBody moved to its own file to stay under the line cap.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
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
1 participant
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
gen_ai.chatspan formessages.create({ stream: true })was only ended by the SDKStream's async iterator. A caller who takes.asResponse()or.withResponse()and readsresponse.bodythemselves never touches that iterator, so nothing ended the span and it was dropped, leaving a barehttp.clientspan for the LLM call. Found in a Flue app, whose@earendil-works/pi-ailayer always calls.asResponse().Root cause: not what the issue says. The result handed to
deferSpanEndis aStream, not a rawResponse, because the orchestrion promise wrapper side-chains.then()on theAPIPromiseand that forcesparse(). So the check passes, the span's fate is handed to aStreamnobody will ever iterate, and it leaks.Nothing on that
Streampoints back at theResponseit was built from, so the fix hooksStream.fromSSEResponse, the one place the two meet. It runs inside the tracedmessages.createcall, sogetActiveSpan()there is the span to link. ItsResponsebody then gets wrapped.Instrumenting the body rather than the
Streamis what makes this simple: every drain path bottoms out inresponse.body, so one accumulator covers the raw-Responsereader, the async iterator,tee()andtoReadableStream().tee()was losing its span the same way, since it callsthis.iterator()directly. The async-iterator instrumentation is gone from the channel path, since keeping both would have needed a handshake to stop them double-counting; the manualinstrumentAnthropicAiClientpath still uses it.Two details that are easy to get wrong and are commented in place: the wrapper takes its reader on the first read rather than at install time, and its stream has a high-water mark of 0. Either one left at the default disturbs the body, which makes
text(),arrayBuffer()andclone()throw on a response the caller hasn't read yet.Verified across
asResponse,withResponse, the iterator,tee,toReadableStreamand an earlybreak: all six now produce identical attributes, including token usage and finish reasons. A streaming call whose body we can't wrap ends atasyncEndwith request attributes instead of hanging.Fixes #24258