feat: return raw bytes from call_runner for non-JSON responses

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, _ = 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(

src/livepeer_gateway/live_runner.py

@@ -26,10 +26,11 @@
 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
-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 +130,9 @@ class LiveRunnerCallResult:
         repr=False,
         compare=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 = ""
 
 
 @dataclass
@@ -740,6 +744,9 @@ 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.
+
+    ``application/json`` and ``+json`` types parse into ``result.data``; anything else
+    (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:
@@ -800,16 +807,27 @@ async def call_runner(
                     resp.status, resp.headers, runner_url, runner, payment_session, session, resp,
                 )
 
-            data = await request_json(
+            body, content_type = await request_data(
                 runner_url,
                 method=method,
                 payload=request_payload,
                 **request_kwargs,
             )
-            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 `content`.
+            is_json = _is_json_content_type(content_type)
+            data: dict[str, Any] = {}
+            if is_json:
+                try:
+                    data = json.loads(body)
+                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,
@@ -819,6 +837,8 @@ 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=None if is_json else body,
+                content_type=content_type,
             )
         except LivepeerHTTPError as e:
             if e.status_code != 402:
@@ -1137,6 +1157,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

tests/test_call_runner_raw.py

@@ -0,0 +1,153 @@
+"""Tests for non-JSON (raw byte) responses in call_runner.
+
+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.content`` 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.content 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.content == FAKE_JPEG
+    assert result.content_type == "image/jpeg"
+    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.content 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.content == 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")
+
+    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") 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():
+    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)