feat: add SSE/chunked streaming to call_runner

[rickstaa]

What

Adds streaming support to call_runner for SSE / chunked responses, on top of the live-runner work in #20.

• http.open_stream() — opens an HTTP request and hands back the live (session, response) without reading the body, for streaming responses. Caller owns both. No total timeout (streams run indefinitely); only connect/first-byte are bounded. Raises LivepeerHTTPError on >= 400.
• LiveRunnerCallStream — returned by call_runner(..., stream=True). Owns the aiohttp session/response, is an async context manager (or call aclose()), and exposes the body via aiter_bytes() / aiter_lines() plus status/headers/content_type/runner.
• Payment — streaming reuses the existing per-call 402 challenge/retry path. open_stream raises before any body, so the payment retry catches it exactly like the non-streaming path.
• Export LiveRunnerCallStream from the package root.

Notes

• Continuous/per-segment payment remains the trickle/lv2v path; this SSE path is one job, one upfront payment.

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on base/target branches other than the default branch.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 9c4cc2e9-fed8-4476-880a-3702a75e3df8

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch rs/live-runner-streaming

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[linear-code]

ENG-110

[j0sh]

This to support SSE with single shot selection requests, right? SSE should already be supported out of the box with reserved sessions. Also I suppose it won't do much for payments until we get single-shot payments figured out, but the selection support is good to have.

Curious, how does this compose with existing SSE clients / libraries? Is there an example of how to use this?

Small naming suggestion, maybe name the param sse rather than stream to more accurately describe the mechanism.

[rickstaa]

This to support SSE with single shot selection requests, right? SSE should already be supported out of the box with reserved sessions. Also I suppose it won't do much for payments until we get single-shot payments figured out, but the selection support is good to have.

Curious, how does this compose with existing SSE clients / libraries? Is there an example of how to use this?

Small naming suggestion, maybe name the param sse rather than stream to more accurately describe the mechanism.

@j0sh thanks for the review. Agreed, go-livepeer already has full SSE support. This PR is only about the call_runner single-shot path: the current logic buffers via request_json, which waits for the body to complete. For a finite response that's just a buffered return; for an SSE endpoint the body never ends, so it blocks until the timeout and errors. stream=True swaps in open_stream (no body deadline) and hands back the live response.

To see this in action, set the gateway SDK branch to main in my vllm example and remove the stream argument from call_runner. It then fails, because the buffering request_json path can't handle the text/event-stream vLLM returns — it tries to JSON-parse the SSE frames and throws (Expecting value: line 1 column 1).

On naming: I kept stream rather than sse because the handle is transport-agnostic, it should work for chunked/NDJSON as well as SSE:

• aiter_bytes() → any byte stream (binary, chunked, SSE).
• aiter_lines() → any newline-delimited stream (NDJSON, SSE, plain text).

Also it matches the stream=True parameter convention from the OpenAI SDK.