feat(live-runner): self-funding metered single-shot calls

examples/text/README.md

@@ -37,3 +37,12 @@ Verify the runner registration with go-livepeer:
 ```
 curl http://localhost:8935/discovery | jq
 ```
+
+Or stream through the SDK client, which self-funds metered calls when a
+remote signer is given (pays the 402 challenge, then keeps paying on a
+cadence for as long as the stream runs):
+
+```sh
+uv run client.py http://localhost:8935/apps/story-runner/app/sse
+uv run client.py https://orch:8935/apps/story-runner/app/sse --signer-url http://localhost:7936
+```

examples/text/client.py

@@ -0,0 +1,48 @@
+"""SDK client for the text stream demo: a self-funding single-shot SSE call.
+
+Offchain (no payments):
+
+    uv run client.py http://localhost:8935/apps/story-runner/app/sse
+
+On-chain: pass a remote signer. call_runner pays the 402 challenge and,
+for metered pricing (hour/seconds or 720p), keeps funding the call on a
+cadence while the stream runs — no payment code needed here.
+
+    uv run client.py https://orch:8935/apps/story-runner/app/sse \
+        --signer-url http://localhost:7936
+"""
+from __future__ import annotations
+
+import argparse
+import asyncio
+
+from livepeer_gateway.live_runner import call_runner
+
+
+async def main() -> None:
+    p = argparse.ArgumentParser(description="Stream a story from a single-shot runner.")
+    p.add_argument("runner_url", help="Runner app endpoint, e.g. .../apps/story-runner/app/sse")
+    p.add_argument("--signer-url", default=None, help="Remote signer URL. Omit for offchain.")
+    p.add_argument(
+        "--payment-unit",
+        default=None,
+        help="hour|seconds|720p|720p-pixel-seconds|fixed. Default: metered (live).",
+    )
+    args = p.parse_args()
+
+    stream = await call_runner(
+        args.runner_url,
+        method="GET",
+        signer_url=args.signer_url,
+        payment_unit=args.payment_unit,
+        stream=True,
+    )
+    async with stream:
+        async for line in stream.aiter_lines():
+            print(line)
+        if stream.released:
+            print("(session released by orchestrator)")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/livepeer_gateway/http.py

@@ -371,6 +371,50 @@ async def get_json(
     return await request_json(url, headers=headers, timeout=timeout)
 
 
+async def post_empty(
+    url: str,
+    *,
+    headers: Optional[dict[str, str]] = None,
+    timeout: float = 5.0,
+) -> None:
+    """
+    POST an empty body to `url` and discard the response.
+
+    Certificate verification is disabled, matching every other HTTP helper in
+    the SDK. Error responses raise the same typed errors as request_json
+    (SignerRefreshRequired for 480, SkipPaymentCycle for 482, LivepeerHTTPError
+    otherwise) so callers can branch on status codes.
+    """
+    try:
+        client_timeout = aiohttp.ClientTimeout(total=timeout)
+        connector = aiohttp.TCPConnector(ssl=False)
+        async with aiohttp.ClientSession(timeout=client_timeout, connector=connector) as session:
+            async with session.post(url, data=b"", headers=headers) as resp:
+                if resp.status >= 400:
+                    raw = await resp.text()
+                    _raise_http_json_error(resp.status, url, raw, dict(resp.headers.items()))
+                await resp.read()
+    except (SignerRefreshRequired, SkipPaymentCycle, LivepeerGatewayError):
+        raise
+    except ConnectionRefusedError as e:
+        raise LivepeerGatewayError(
+            f"HTTP empty POST error: connection refused (is the server running? is the host/port correct?) (url={url})"
+        ) from e
+    except getattr(aiohttp, "ClientConnectorError", ()) as e:
+        os_error = getattr(e, "os_error", None)
+        if isinstance(os_error, ConnectionRefusedError):
+            raise LivepeerGatewayError(
+                f"HTTP empty POST error: connection refused (is the server running? is the host/port correct?) (url={url})"
+            ) from e
+        raise LivepeerGatewayError(
+            f"HTTP empty POST error: failed to reach endpoint: {getattr(e, 'message', e)} (url={url})"
+        ) from e
+    except (aiohttp.ClientError, asyncio.TimeoutError) as e:
+        raise LivepeerGatewayError(
+            f"HTTP empty POST error: failed to reach endpoint: {getattr(e, 'message', e)} (url={url})"
+        ) from e
+
+
 def _parse_http_url(url: str, *, context: str = "URL") -> ParseResult:
     """
     Normalize a URL for HTTP(S) endpoints.