feat(adapters): add streaming support to WatsonxOrchestrateAgent

[sundowatch]

Summary

WatsonxOrchestrateAgent always issued a blocking, non-streaming HTTP request to the WatsonX Orchestrate API (stream=False), even though the API fully supports Server-Sent Events (SSE) streaming. This forced the framework to wait for the entire response before returning anything, adding unnecessary latency.

This PR implements streaming for WatsonxOrchestrateAgent.run() and fixes the related finish_reason hardcoding in WatsonxOrchestrateServerAgent.run().

Changes

agents/agent.py — streaming support
Replaces the single client.post() call with client.stream() + SSE parsing:

• Sends stream=True in the request body
• Iterates aiter_lines() on the streaming response
• Parses each data: {...} SSE line; stops on [DONE] sentinel
• Accumulates content deltas and captures finish_reason from the final chunk
• Assembles a ChatCompletionResponse from the collected data, preserving the existing WatsonxOrchestrateAgentOutput interface — no breaking changes

serve/agent.py — fix hardcoded finish_reason
WatsonxOrchestrateServerAgent.run() was hardcoding finish_reason="stop" for all responses regardless of what the agent actually returned. Now it reads the actual finish_reason from the response's raw field when available (populated by WatsonxOrchestrateAgent after this change), falling back to "stop" for other agent types.

Test plan

• Syntax and ruff lint pass on both modified files
• No changes to the public interface (WatsonxOrchestrateAgentOutput, WatsonxOrchestrateAgent constructor)
• Manual verification requires a WatsonX Orchestrate instance — SSE parsing follows the standard OpenAI-compatible format already used elsewhere in the serve layer

[gemini-code-assist]

Code Review

This pull request introduces streaming support for the Watsonx Orchestrate agent adapter by transitioning to a streaming client, parsing Server-Sent Events (SSE), and reconstructing the chat completion response. It also updates the serving layer to dynamically extract the finish_reason from raw response choices and adds comprehensive unit tests for StreamToolCallMiddleware and GlobalTrajectoryMiddleware. The review feedback recommends enhancing robustness by handling SSE lines without a space after the data: prefix and adding defensive type checks when parsing JSON chunks and raw response attributes to prevent potential runtime crashes.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[sundowatch]

Added unit tests for the SSE chat-completion parsing in cc027efa: I pulled the accumulation out into a small pure helper and covered multi-chunk content, the [DONE] sentinel, the optional space after data:, and the defensive handling for malformed/unexpected payloads. The goal is to make the streaming path verifiable in CI without a live WatsonX Orchestrate instance. Happy to tweak the approach if you'd prefer it structured differently.

[Tomas2D]

Thanks @sundowatch — the code is correct, the SSE parser is nicely defensive, and CI is green. One substantive note: this doesn't actually stream to the caller. run() collects the entire stream (lines = [line async for line in streaming_response.aiter_lines()]) before parsing, so it still blocks until the full response arrives — the framework's streaming convention is per-token emitter events (e.g. new_token/partial_update), which aren't emitted here. So this is really "consume the SSE endpoint" rather than "add streaming"; consider wiring incremental emitter events or reframing the title/description.

Minor: consider await streaming_response.aread() before raise_for_status() so server error bodies aren't lost, and an integration test for the run() path (only the pure parser is currently tested). The serve/agent.py finish_reason fix looks good.

[sundowatch]

Thanks for the detailed review, @Tomas2D — you're right that the previous version only consumed the SSE endpoint without streaming to the caller. I've reworked it in ca8d1a5d:

• Real incremental streaming: run() now parses each chunk as it arrives and emits per-token update events (WatsonxOrchestrateAgentUpdateEvent carrying the delta and the accumulated content), registered on the agent's emitter — so it follows the framework's emitter-based streaming convention instead of blocking until the full response is buffered. The fully accumulated result is still returned as before.
• Error bodies: it now reads the body when is_error before raise_for_status(), so the server's error detail is preserved — without consuming the body on the success (streaming) path.
• Tests: kept the pure per-chunk parser unit tests and added a run()-path integration test with a mocked stream that asserts the incremental update events actually fire per token (not one blob at the end) and that the final content / [DONE] handling is correct — all runnable in CI without a live WatsonX Orchestrate instance.

So the title now genuinely reflects streaming. Happy to rename the update event (I followed the a2a adapter's convention) or adjust anything else you'd prefer.