fix(server-utils): Instrument LangGraph stream executions

[psh4607]

LangGraph CompiledGraph.stream() now creates the same gen_ai.invoke_agent parent span as invoke() and keeps it active until stream consumption finishes. The shared wrapper is used by manual server-runtime instrumentation and diagnostics-channel auto-instrumentation.

Root cause

The compile and create-agent wrappers only patched invoke(). stream() returned an async iterable without an agent span, so callback-created child spans became roots. Ending a span when stream() resolves would also be too early because graph work continues while the stream is consumed.

The implementation preserves the original stream object, observes async-iterator and ReadableStream consumption paths, forwards chunks unchanged, records accumulated response attributes when enabled, and ends the span on completion, cancellation, or error. It also suppresses LangGraph's internal invoke() to stream() call per compiled graph so existing invoke instrumentation does not emit duplicate agent spans.

JS-1858

Fixes #19626

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit 794a336. Configure here.}

[github-actions]

👋 @isaacs, @JPeer264, @getsentry/team-javascript-sdks — Please review this PR when you get a chance!

[sentry]

Bug: A recursive call to graph.stream() from a tool within a graph.invoke() execution will have its tracing suppressed, leading to unmonitored execution.
_{Severity: MEDIUM}

Suggested Fix

The suppression logic should be more specific to differentiate between a true nested invoke and a recursive stream call. Consider using a different flag for stream vs. invoke, or clearing the LANGGRAPH_INVOKE_ACTIVE flag before executing tool code to prevent it from being inherited by the tool's execution scope.

Prompt for AI Agent

Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.

Location: packages/server-utils/src/ai/langgraph/index.ts#L157-L162

Potential issue: The `LANGGRAPH_INVOKE_ACTIVE` flag, set during a `graph.invoke()` call,
is inherited by all descendant async scopes. If a tool executed within the `invoke()`
call makes a recursive call to `graph.stream()` on the same graph instance, the
suppression check will incorrectly identify this as a nested `invoke` and suppress
instrumentation for the `stream()` call. This results in the entire `stream()` execution
being unmonitored, losing all tracing data for that operation.

[RulaKhaled]

This is valid, the token stays on the scope for the whole invoke(), not just the internal stream() call it's meant to cover, so any nested stream() on this same compiled graph runs uninstrumented for the rest of the invoke.

something like this would work:

getCurrentScope().setSDKProcessingMetadata({ [LANGGRAPH_INVOKE_ACTIVE]: undefined });
return Reflect.apply(target, thisArg, args);

[sentry]

Bug: A span leak occurs if a stream returned by graph.stream() is created but never iterated, as the span-closing logic is in a finally block that never runs.
_{Severity: MEDIUM}

Suggested Fix

Ensure the span is closed even if the stream is not consumed. One approach is to wrap the stream in a proxy that closes the span upon garbage collection using FinalizationRegistry. Alternatively, the startSpanManual call could be moved inside the async generator so the span is only created when iteration begins.

Prompt for AI Agent

Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.

Location: packages/server-utils/src/ai/langgraph/index.ts#L274

Potential issue: When `graph.stream()` is called, it creates a manual span that is only
closed when the returned stream is consumed. If the stream is created but never iterated
(e.g., due to an error after creation but before consumption), the `finally` block that
closes the span is never executed. This occurs because the async generator body is not
entered until the first `.next()` call. An un-iterated stream that is garbage collected
will leak its associated span, which will remain open indefinitely.

[RulaKhaled]

Thanks for digging into this! the stream lifecycle work is in good shape. i just kicked off the CI workflow, meanwhile there are a couple of comments to look at

[RulaKhaled]

This is valid, the token stays on the scope for the whole invoke(), not just the internal stream() call it's meant to cover, so any nested stream() on this same compiled graph runs uninstrumented for the rest of the invoke.

something like this would work:

getCurrentScope().setSDKProcessingMetadata({ [LANGGRAPH_INVOKE_ACTIVE]: undefined });
return Reflect.apply(target, thisArg, args);

[RulaKhaled]

nothing currently covers the orphaned-child-span symptom from the issue. Can you add an agent-stream-scenario.mjs (like agent-scenario.mjs, but streaming) asserting the chat span lands under the agent span?

[RulaKhaled]

m: streamMode: 'messages' emits [messageChunk, metadata] (a tuple whose first element is a message object) falls through here, could we cover this as well?

[RulaKhaled]

lifecycle.fail() calls treat a caller TypeError as a stream failure, I hit this with a live getReader(), the stray pipeTo rejects, the span ends as internal_error. Could we guard the preconditions up front if (stream.locked || destination.locked) return Promise.reject(new TypeError(...)) rejecting rather than throwing? Maybe add a test that a mid consumption pipeTo TypeError doesn't touch span status/end?