From ef373d5ec72c7ea4c5f912ed8bf53b1cc4a264ca Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Wed, 29 Jul 2026 20:21:02 +0200 Subject: [PATCH 1/4] feat: return raw bytes from call_runner for non-JSON responses Co-Authored-By: Claude Fable 5 --- src/livepeer_gateway/http.py | 53 +++++++++++--- src/livepeer_gateway/live_runner.py | 30 +++++++- tests/test_call_runner_raw.py | 110 ++++++++++++++++++++++++++++ 3 files changed, 181 insertions(+), 12 deletions(-) create mode 100644 tests/test_call_runner_raw.py diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index 043d88a..17738dd 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -240,20 +240,23 @@ def get_json_sync( return request_json_sync(url, headers=headers, timeout=timeout) -async def request_json( +async def request_data( url: str, *, method: Optional[str] = None, payload: Optional[dict[str, Any]] = None, headers: Optional[dict[str, str]] = None, timeout: float = 5.0, -) -> Any: +) -> tuple[bytes, str]: """ - Make an async JSON HTTP request and parse the JSON response. + Make an async JSON-payload HTTP request and return the raw response body. + + Returns ``(body, content_type)`` without assuming the response is JSON; + request semantics and error mapping match request_json. If method is None, defaults to POST when payload is provided, otherwise GET. - Raises LivepeerGatewayError on HTTP/network/JSON parsing errors. + Raises LivepeerGatewayError on HTTP/network errors. """ resolved_method, req_headers, body = _json_request_parts( url, @@ -267,14 +270,14 @@ async def request_json( connector = aiohttp.TCPConnector(ssl=False) async with aiohttp.ClientSession(timeout=client_timeout, connector=connector) as session: async with session.request(resolved_method, url, data=body, headers=req_headers) as resp: - raw = await resp.text() + raw = await resp.read() + content_type = resp.content_type or "" if resp.status >= 400: - _raise_http_json_error(resp.status, url, raw, dict(resp.headers.items())) - data: Any = json.loads(raw) + _raise_http_json_error( + resp.status, url, raw.decode(errors="replace"), dict(resp.headers.items()) + ) except (SignerRefreshRequired, SkipPaymentCycle, LivepeerGatewayError): raise - except json.JSONDecodeError as e: - raise LivepeerGatewayError(f"HTTP JSON error: endpoint did not return valid JSON: {e} (url={url})") from e except ConnectionRefusedError as e: raise LivepeerGatewayError( f"HTTP JSON error: connection refused (is the server running? is the host/port correct?) (url={url})" @@ -297,7 +300,37 @@ async def request_json( f"HTTP JSON error: unexpected error: {e.__class__.__name__}: {e} (url={url})" ) from e - return data + return raw, content_type + + +async def request_json( + url: str, + *, + method: Optional[str] = None, + payload: Optional[dict[str, Any]] = None, + headers: Optional[dict[str, str]] = None, + timeout: float = 5.0, +) -> Any: + """ + Make an async JSON HTTP request and parse the JSON response. + + If method is None, defaults to POST when payload is provided, otherwise GET. + + Raises LivepeerGatewayError on HTTP/network/JSON parsing errors. + """ + raw, _content_type = await request_data( + url, + method=method, + payload=payload, + headers=headers, + timeout=timeout, + ) + try: + return json.loads(raw) + except json.JSONDecodeError as e: + raise LivepeerGatewayError( + f"HTTP JSON error: endpoint did not return valid JSON: {e} (url={url})" + ) from e async def open_stream( diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index fbe3010..e94325b 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -29,7 +29,7 @@ from .channel_reader import ChannelReader from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired -from .http import open_stream, post_json, request_json +from .http import open_stream, post_json, request_data, request_json from .remote_signer import ( GetPaymentResponse, LivePaymentSession, @@ -129,6 +129,10 @@ class LiveRunnerCallResult: repr=False, compare=False, ) + # Non-JSON responses (an image, say) arrive unparsed: `raw` holds the body + # bytes and `content_type` its media type, while `data` stays empty. + raw: Optional[bytes] = field(default=None, repr=False) + content_type: str = "" @dataclass @@ -740,6 +744,10 @@ async def call_runner( With ``signer_url`` set, payment is automatic and **per call**: a 402 challenge is paid via the signer and retried (up to ``max_payment_challenge_retries``), one job, one upfront payment. Raises ``LivepeerHTTPError`` on non-402 errors. + + JSON responses (by content type) are parsed into ``result.data``; anything else + (an image, say) is returned unparsed in ``result.raw`` with ``result.content_type`` + set. """ runner_url = runner_url.strip() or (runner.url.strip() if runner is not None else "") if not runner_url: @@ -800,12 +808,29 @@ async def call_runner( resp.status, resp.headers, runner_url, runner, payment_session, session, resp, ) - data = await request_json( + raw, content_type = await request_data( runner_url, method=method, payload=request_payload, **request_kwargs, ) + if "json" not in content_type.lower(): + # Non-JSON body (an image, say): hand back the bytes unparsed. + return LiveRunnerCallResult( + {}, + runner_url=runner_url, + runner=runner, + session_id=session_id, + payment_session=None if payment_type == "fixed" else payment_session, + raw=raw, + content_type=content_type, + ) + try: + data = json.loads(raw) + except json.JSONDecodeError as e: + raise LivepeerGatewayError( + f"HTTP JSON error: endpoint did not return valid JSON: {e} (url={runner_url})" + ) from e if not isinstance(data, dict): raise LivepeerGatewayError( f"Live runner call expected JSON object, got {type(data).__name__}" @@ -819,6 +844,7 @@ async def call_runner( or (data["session_id"].strip() if isinstance(data.get("session_id"), str) else "") ), payment_session=None if payment_type == "fixed" else payment_session, + content_type=content_type, ) except LivepeerHTTPError as e: if e.status_code != 402: diff --git a/tests/test_call_runner_raw.py b/tests/test_call_runner_raw.py new file mode 100644 index 0000000..545a859 --- /dev/null +++ b/tests/test_call_runner_raw.py @@ -0,0 +1,110 @@ +"""Tests for non-JSON (raw byte) responses in call_runner. + +JSON responses (by content type) keep today's behavior: parsed into +``result.data``, strict about being an object. Any other content type returns +the body unparsed in ``result.raw`` with ``result.content_type`` set. +""" + +from __future__ import annotations + +import asyncio + +import pytest +from aiohttp import web + +from livepeer_gateway.errors import LivepeerGatewayError, LivepeerHTTPError +from livepeer_gateway.live_runner import call_runner + +FAKE_JPEG = b"\xff\xd8\xff\xe0" + b"jpeg-bytes" * 100 + + +def _run(app: web.Application, scenario): + """Serve `app` on an ephemeral port and run `scenario(base_url)`.""" + + async def main(): + runner = web.AppRunner(app) + await runner.setup() + site = web.TCPSite(runner, "127.0.0.1", 0) + await site.start() + port = site._server.sockets[0].getsockname()[1] + try: + return await scenario(f"http://127.0.0.1:{port}") + finally: + await runner.cleanup() + + return asyncio.run(main()) + + +def test_json_response_unchanged(): + async def handler(request): + return web.json_response({"message": "hello", "session_id": " s1 "}) + + app = web.Application() + app.router.add_post("/call", handler) + + async def scenario(base): + return await call_runner(f"{base}/call", payload={"x": 1}) + + result = _run(app, scenario) + assert result.data == {"message": "hello", "session_id": " s1 "} + assert result.session_id == "s1" + assert result.raw is None + assert result.content_type == "application/json" + + +def test_binary_response_returns_raw(): + async def handler(request): + return web.Response(body=FAKE_JPEG, content_type="image/jpeg") + + app = web.Application() + app.router.add_post("/img", handler) + + async def scenario(base): + return await call_runner(f"{base}/img", payload={"prompt": "x"}) + + result = _run(app, scenario) + assert result.raw == FAKE_JPEG + assert result.content_type == "image/jpeg" + assert result.data == {} + + +def test_invalid_json_with_json_content_type_raises(): + async def handler(request): + return web.Response(text="not json", content_type="application/json") + + app = web.Application() + app.router.add_post("/bad", handler) + + async def scenario(base): + return await call_runner(f"{base}/bad", payload={}) + + with pytest.raises(LivepeerGatewayError, match="did not return valid JSON"): + _run(app, scenario) + + +def test_json_array_still_rejected(): + async def handler(request): + return web.json_response([1, 2, 3]) + + app = web.Application() + app.router.add_post("/arr", handler) + + async def scenario(base): + return await call_runner(f"{base}/arr", payload={}) + + with pytest.raises(LivepeerGatewayError, match="expected JSON object"): + _run(app, scenario) + + +def test_http_error_still_raises_with_binary_endpoint(): + async def handler(request): + return web.Response(status=404, text="nope") + + app = web.Application() + app.router.add_post("/missing", handler) + + async def scenario(base): + return await call_runner(f"{base}/missing", payload={}) + + with pytest.raises(LivepeerHTTPError): + _run(app, scenario) From 123cb656cdb88e7a04bc552fec0a3c8dc73d6d51 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 30 Jul 2026 11:11:13 +0200 Subject: [PATCH 2/4] refactor: match single-document JSON by media subtype MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Content-type detection was a substring test (`"json" in content_type`), which is right for `+json` types only by accident and wrong for multi-document formats: `application/jsonl` (this repo's own trickle channel default), `application/x-ndjson`, and `application/json-seq` all matched and then failed in json.loads, so a working runner fronting a streaming API got "did not return valid JSON" instead of its bytes — the same bug class this branch set out to fix. Match on the media subtype instead, via aiohttp's own mimetype parser: `application/json` or the RFC 6839 `+json` structured suffix. Vendor types (`application/vnd.acme.v1+json`) keep parsing without being listed; multi-document bodies fall through to `raw`. Only four classifications change, all previously raising. Also fold the raw early return into the single existing return, so `payment_session=None if payment_type == "fixed" else payment_session` stays in one place — payment-type handling here has been reverted twice and there is no paid-binary test to catch the two sites drifting. The `session_id` expression reduces to the early return's behavior when `data` is empty, so this is equivalent. Add `content_type` to the JSON parse error: it is the fact that routed the response into parsing, and without it a misclassification is undebuggable from the message alone. Co-Authored-By: Claude Opus 5 (1M context) --- src/livepeer_gateway/http.py | 2 +- src/livepeer_gateway/live_runner.py | 48 ++++++++++++++-------------- tests/test_call_runner_raw.py | 49 +++++++++++++++++++++++++++-- 3 files changed, 71 insertions(+), 28 deletions(-) diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index 17738dd..9448530 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -318,7 +318,7 @@ async def request_json( Raises LivepeerGatewayError on HTTP/network/JSON parsing errors. """ - raw, _content_type = await request_data( + raw, _ = await request_data( url, method=method, payload=payload, diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index e94325b..6324a46 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -26,6 +26,7 @@ from urllib.parse import quote, urlparse, urlunparse import aiohttp +from aiohttp.helpers import parse_mimetype from .channel_reader import ChannelReader from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired @@ -745,9 +746,8 @@ async def call_runner( paid via the signer and retried (up to ``max_payment_challenge_retries``), one job, one upfront payment. Raises ``LivepeerHTTPError`` on non-402 errors. - JSON responses (by content type) are parsed into ``result.data``; anything else - (an image, say) is returned unparsed in ``result.raw`` with ``result.content_type`` - set. + ``application/json`` and ``+json`` types parse into ``result.data``; anything else + (an image, ndjson) comes back unparsed in ``result.raw`` + ``result.content_type``. """ runner_url = runner_url.strip() or (runner.url.strip() if runner is not None else "") if not runner_url: @@ -814,27 +814,21 @@ async def call_runner( payload=request_payload, **request_kwargs, ) - if "json" not in content_type.lower(): - # Non-JSON body (an image, say): hand back the bytes unparsed. - return LiveRunnerCallResult( - {}, - runner_url=runner_url, - runner=runner, - session_id=session_id, - payment_session=None if payment_type == "fixed" else payment_session, - raw=raw, - content_type=content_type, - ) - try: - data = json.loads(raw) - except json.JSONDecodeError as e: - raise LivepeerGatewayError( - f"HTTP JSON error: endpoint did not return valid JSON: {e} (url={runner_url})" - ) from e - if not isinstance(data, dict): - raise LivepeerGatewayError( - f"Live runner call expected JSON object, got {type(data).__name__}" - ) + # Non-JSON bodies (an image, ndjson) are handed back unparsed in `raw`. + is_json = _is_json_content_type(content_type) + data: dict[str, Any] = {} + if is_json: + try: + data = json.loads(raw) + except json.JSONDecodeError as e: + raise LivepeerGatewayError( + f"HTTP JSON error: endpoint did not return valid JSON: {e} " + f"(url={runner_url}, content_type={content_type})" + ) from e + if not isinstance(data, dict): + raise LivepeerGatewayError( + f"Live runner call expected JSON object, got {type(data).__name__}" + ) return LiveRunnerCallResult( data, runner_url=runner_url, @@ -844,6 +838,7 @@ async def call_runner( or (data["session_id"].strip() if isinstance(data.get("session_id"), str) else "") ), payment_session=None if payment_type == "fixed" else payment_session, + raw=None if is_json else raw, content_type=content_type, ) except LivepeerHTTPError as e: @@ -1163,6 +1158,11 @@ def _validate_trickle_channel_requests(channels: list[LiveRunnerTrickleChannelRe raise TypeError("trickle channel mime_type must be str") +def _is_json_content_type(content_type: str) -> bool: + mime = parse_mimetype(content_type) + return mime.subtype == "json" or mime.suffix == "json" + + def _is_trickle_channel_response(value: object) -> bool: if not isinstance(value, dict): return False diff --git a/tests/test_call_runner_raw.py b/tests/test_call_runner_raw.py index 545a859..5db6b56 100644 --- a/tests/test_call_runner_raw.py +++ b/tests/test_call_runner_raw.py @@ -1,7 +1,8 @@ """Tests for non-JSON (raw byte) responses in call_runner. -JSON responses (by content type) keep today's behavior: parsed into -``result.data``, strict about being an object. Any other content type returns +Single-document JSON responses (``application/json`` or an RFC 6839 ``+json`` +suffix) keep today's behavior: parsed into ``result.data``, strict about being an +object. Anything else — binary, or a multi-document format like ndjson — returns the body unparsed in ``result.raw`` with ``result.content_type`` set. """ @@ -68,6 +69,46 @@ async def scenario(base): assert result.data == {} +def test_json_suffix_content_type_is_parsed(): + """RFC 6839 ``+json`` types are single JSON documents, so they parse.""" + + async def handler(request): + return web.Response( + text='{"message": "hello"}', content_type="application/vnd.acme.v1+json" + ) + + app = web.Application() + app.router.add_post("/vnd", handler) + + async def scenario(base): + return await call_runner(f"{base}/vnd", payload={}) + + result = _run(app, scenario) + assert result.data == {"message": "hello"} + assert result.raw is None + assert result.content_type == "application/vnd.acme.v1+json" + + +def test_ndjson_returns_raw(): + """Multi-document formats json.loads can't parse come back as bytes.""" + + body = b'{"token": "Hello"}\n{"token": " world"}\n' + + async def handler(request): + return web.Response(body=body, content_type="application/x-ndjson") + + app = web.Application() + app.router.add_post("/ndjson", handler) + + async def scenario(base): + return await call_runner(f"{base}/ndjson", payload={}) + + result = _run(app, scenario) + assert result.raw == body + assert result.content_type == "application/x-ndjson" + assert result.data == {} + + def test_invalid_json_with_json_content_type_raises(): async def handler(request): return web.Response(text="not json", content_type="application/json") @@ -78,8 +119,10 @@ async def handler(request): async def scenario(base): return await call_runner(f"{base}/bad", payload={}) - with pytest.raises(LivepeerGatewayError, match="did not return valid JSON"): + with pytest.raises(LivepeerGatewayError, match="did not return valid JSON") as excinfo: _run(app, scenario) + # The content type is in the message: it is what routed us into parsing. + assert "content_type=application/json" in str(excinfo.value) def test_json_array_still_rejected(): From f57625f27aae6f7a1b17a08c2cdb6ab076b6c13a Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 30 Jul 2026 11:20:57 +0200 Subject: [PATCH 3/4] docs: trim result field comment and helper docstring `content_type` is self-documenting from the field name; the invariant worth stating is that a populated `raw` means an empty `data`. The subtype helper matches its bare neighbors in this module. Co-Authored-By: Claude Opus 5 (1M context) --- src/livepeer_gateway/live_runner.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 6324a46..aedda0a 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -130,8 +130,7 @@ class LiveRunnerCallResult: repr=False, compare=False, ) - # Non-JSON responses (an image, say) arrive unparsed: `raw` holds the body - # bytes and `content_type` its media type, while `data` stays empty. + # Non-JSON responses (an image, say) arrive unparsed in `raw`; `data` stays empty. raw: Optional[bytes] = field(default=None, repr=False) content_type: str = "" From f42f9da4dff691d26d22003f88d3e7d8779c4cad Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 30 Jul 2026 12:09:01 +0200 Subject: [PATCH 4/4] refactor: name the non-JSON body field `content` MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `raw` already means "the original JSON dict, un-normalized" on three dataclasses in this SDK — LiveRunnerSessionEvent.raw, LiveRunnerInstance.raw, LiveVideoToVideo.raw — and examples dump it with json.dumps(x.raw). A `raw: Optional[bytes]` on LiveRunnerCallResult overloads the name to mean the opposite, sharply so on one expression chain: `result.raw` (bytes) sitting one dot from `result.runner.raw` (dict). `content` matches its sibling `content_type` and the ecosystem convention for a response body as bytes (requests/httpx `.content`) — where `.raw` instead means an unread stream object, so the old name actively misled. Nothing consumes the field yet (runner-app-examples#45 and api-proxy both use the streaming path), so this is free now and permanent once released. Semantics unchanged: still `Optional[bytes] = None`, since `b""` is a valid empty body and `None` is the only unambiguous "this was JSON" sentinel. Also renames the local `raw` to `body` — it holds bytes, while `raw` in this codebase reads as a dict. Co-Authored-By: Claude Opus 5 (1M context) --- src/livepeer_gateway/live_runner.py | 14 +++++++------- tests/test_call_runner_raw.py | 10 +++++----- 2 files changed, 12 insertions(+), 12 deletions(-) diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index aedda0a..fe366c1 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -130,8 +130,8 @@ class LiveRunnerCallResult: repr=False, compare=False, ) - # Non-JSON responses (an image, say) arrive unparsed in `raw`; `data` stays empty. - raw: Optional[bytes] = field(default=None, repr=False) + # Non-JSON responses (an image, say) arrive unparsed in `content`; `data` stays empty. + content: Optional[bytes] = field(default=None, repr=False) content_type: str = "" @@ -746,7 +746,7 @@ async def call_runner( one upfront payment. Raises ``LivepeerHTTPError`` on non-402 errors. ``application/json`` and ``+json`` types parse into ``result.data``; anything else - (an image, ndjson) comes back unparsed in ``result.raw`` + ``result.content_type``. + (an image, ndjson) comes back unparsed in ``result.content`` + ``result.content_type``. """ runner_url = runner_url.strip() or (runner.url.strip() if runner is not None else "") if not runner_url: @@ -807,18 +807,18 @@ async def call_runner( resp.status, resp.headers, runner_url, runner, payment_session, session, resp, ) - raw, content_type = await request_data( + body, content_type = await request_data( runner_url, method=method, payload=request_payload, **request_kwargs, ) - # Non-JSON bodies (an image, ndjson) are handed back unparsed in `raw`. + # Non-JSON bodies (an image, ndjson) are handed back unparsed in `content`. is_json = _is_json_content_type(content_type) data: dict[str, Any] = {} if is_json: try: - data = json.loads(raw) + data = json.loads(body) except json.JSONDecodeError as e: raise LivepeerGatewayError( f"HTTP JSON error: endpoint did not return valid JSON: {e} " @@ -837,7 +837,7 @@ async def call_runner( or (data["session_id"].strip() if isinstance(data.get("session_id"), str) else "") ), payment_session=None if payment_type == "fixed" else payment_session, - raw=None if is_json else raw, + content=None if is_json else body, content_type=content_type, ) except LivepeerHTTPError as e: diff --git a/tests/test_call_runner_raw.py b/tests/test_call_runner_raw.py index 5db6b56..8359300 100644 --- a/tests/test_call_runner_raw.py +++ b/tests/test_call_runner_raw.py @@ -3,7 +3,7 @@ Single-document JSON responses (``application/json`` or an RFC 6839 ``+json`` suffix) keep today's behavior: parsed into ``result.data``, strict about being an object. Anything else — binary, or a multi-document format like ndjson — returns -the body unparsed in ``result.raw`` with ``result.content_type`` set. +the body unparsed in ``result.content`` with ``result.content_type`` set. """ from __future__ import annotations @@ -49,7 +49,7 @@ async def scenario(base): result = _run(app, scenario) assert result.data == {"message": "hello", "session_id": " s1 "} assert result.session_id == "s1" - assert result.raw is None + assert result.content is None assert result.content_type == "application/json" @@ -64,7 +64,7 @@ async def scenario(base): return await call_runner(f"{base}/img", payload={"prompt": "x"}) result = _run(app, scenario) - assert result.raw == FAKE_JPEG + assert result.content == FAKE_JPEG assert result.content_type == "image/jpeg" assert result.data == {} @@ -85,7 +85,7 @@ async def scenario(base): result = _run(app, scenario) assert result.data == {"message": "hello"} - assert result.raw is None + assert result.content is None assert result.content_type == "application/vnd.acme.v1+json" @@ -104,7 +104,7 @@ async def scenario(base): return await call_runner(f"{base}/ndjson", payload={}) result = _run(app, scenario) - assert result.raw == body + assert result.content == body assert result.content_type == "application/x-ndjson" assert result.data == {}