From 42dbb91b19231e7f3b5ebf603a6c9da6c01c4a71 Mon Sep 17 00:00:00 2001
From: Rick Staa <rick.staa@outlook.com>
Date: Thu, 30 Jul 2026 12:40:47 +0200
Subject: [PATCH 01/12] feat(payments): give LivePaymentSession an interval
 payment loop

Metered live-runner pricing debits a prepaid balance on the
orchestrator's -livePaymentInterval and drops the session on the first
tick it cannot cover, so every metered caller needs the same loop.
Put it on the payment session itself, where both the single-shot and
the reserved-session paths already hold one:

- post_empty moves into http.py (Josh's suggestion on #31) so the
  payment POST disables cert verification like the rest of the SDK and
  raises typed status errors; send_payment routes through it and takes
  a payment_url for the session-scoped endpoint, which 404s once the
  session is gone instead of blind-crediting a balance
- run_payments pays on an interval, returns True when the orchestrator
  reports the session released, and stops on the other terminal
  rejections (409 fixed price, 403 mismatch) rather than minting
  tickets no one will honor

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 src/livepeer_gateway/http.py          |  44 +++++++++++
 src/livepeer_gateway/remote_signer.py | 106 ++++++++++++++++++--------
 2 files changed, 118 insertions(+), 32 deletions(-)

diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py
index 043d88a..db4057b 100644
--- a/src/livepeer_gateway/http.py
+++ b/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.
diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py
index fc94463..d21e54c 100644
--- a/src/livepeer_gateway/remote_signer.py
+++ b/src/livepeer_gateway/remote_signer.py
@@ -12,13 +12,22 @@
 from urllib.error import HTTPError, URLError
 from urllib.request import Request, urlopen
 
-import aiohttp
-
 from . import lp_rpc_pb2
 from .async_cache import async_lru_cache
-from .errors import LivepeerGatewayError, PaymentError, SignerRefreshRequired
+from .errors import (
+    LivepeerGatewayError,
+    LivepeerHTTPError,
+    PaymentError,
+    SignerRefreshRequired,
+    SkipPaymentCycle,
+)
 _LOG = logging.getLogger(__name__)
 
+# Client payment cadence. The orchestrator debits metered sessions every
+# -livePaymentInterval (5s by default) and drops the session on the first tick
+# it cannot cover, so pay comfortably ahead of it.
+PAYMENT_INTERVAL_S = 3.0
+
 @dataclass(frozen=True)
 class GetPaymentResponse:
     payment: str
@@ -240,44 +249,77 @@ async def get_payment(self) -> GetPaymentResponse:
                 await self._refresh_payment_params(orchestrator_url)
                 attempts += 1
 
-    async def send_payment(self, orchestrator_url: Optional[str] = None) -> None:
+    async def send_payment(
+        self,
+        orchestrator_url: Optional[str] = None,
+        *,
+        payment_url: Optional[str] = None,
+    ) -> None:
+        """Generate a payment and POST it to the orchestrator.
+
+        ``payment_url`` targets a specific endpoint, such as the session-scoped
+        one which 404s once the session is released. Without it the payment
+        goes to the orchestrator's generic ``/payment`` endpoint, which credits
+        the payer balance blindly and cannot report a dead session.
+
+        Raises LivepeerHTTPError on error responses so callers can branch on
+        the status code, and SkipPaymentCycle when the signer gates the cycle.
+        """
         if not self._signer_url:
             return
 
-        target = orchestrator_url or self._orchestrator_url
-        if not target:
-            raise PaymentError("orchestrator_url is required before sending payment")
+        from .http import _http_origin, post_empty
 
-        from .http import _extract_error_message_from_body, _http_origin
+        if payment_url:
+            url = payment_url
+        else:
+            target = orchestrator_url or self._orchestrator_url
+            if not target:
+                raise PaymentError("orchestrator_url is required before sending payment")
+            url = f"{_http_origin(target)}/payment"
 
         payment = await self.get_payment()
-        url = f"{_http_origin(target)}/payment"
         headers = {
             "Livepeer-Payment": payment.payment,
-            "Livepeer-Segment": payment.seg_creds,
+            "Livepeer-Segment": payment.seg_creds or "",
         }
-        try:
-            timeout = aiohttp.ClientTimeout(total=5.0)
-            async with aiohttp.ClientSession(timeout=timeout) as session:
-                async with session.post(url, data=b"", headers=headers) as resp:
-                    if resp.status >= 400:
-                        body = await resp.text()
-                        message = _extract_error_message_from_body(body)
-                        body_part = f"; body={message!r}" if message else ""
-                        raise PaymentError(
-                            f"HTTP payment error: HTTP {resp.status} from endpoint (url={url}){body_part}"
-                        )
-                    await resp.read()
-        except PaymentError:
-            raise
-        except getattr(aiohttp, "ClientConnectorError", ()) as e:
-            raise PaymentError(
-                f"HTTP payment error: failed to reach endpoint: {getattr(e, 'message', e)} (url={url})"
-            ) from e
-        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
-            raise PaymentError(
-                f"HTTP payment error: failed to reach endpoint: {getattr(e, 'message', e)} (url={url})"
-            ) from e
+        await post_empty(url, headers=headers, timeout=5.0)
+
+    async def run_payments(
+        self,
+        *,
+        payment_url: Optional[str] = None,
+        interval_s: float = PAYMENT_INTERVAL_S,
+    ) -> bool:
+        """Keep a metered session funded until cancelled or the session ends.
+
+        Returns True when the orchestrator reports the session gone, so the
+        owner can surface it as released; returns False for the other terminal
+        rejections. Cancel the task to stop funding.
+
+        The caller pays upfront before starting this loop, so the first
+        follow-up waits one interval. A payment covers the time since the
+        previous one, so transient failures are retried rather than fatal:
+        the next payment settles the arrears.
+        """
+        while True:
+            await asyncio.sleep(interval_s)
+            try:
+                await self.send_payment(payment_url=payment_url)
+            except asyncio.CancelledError:
+                raise
+            except SkipPaymentCycle as e:
+                _LOG.debug("Payment loop skipped cycle: %s", e)
+            except LivepeerHTTPError as e:
+                # 404 session released, 409 fixed price, 403 session/payment
+                # mismatch: all terminal, and paying on would mint tickets the
+                # orchestrator will never honor.
+                if 400 <= e.status_code < 500 and e.status_code not in (408, 429):
+                    _LOG.info("Payment loop stopping (HTTP %d): %s", e.status_code, e)
+                    return e.status_code == 404
+                _LOG.warning("Payment failed; retrying next cycle: %s", e)
+            except Exception as e:
+                _LOG.warning("Payment failed; retrying next cycle: %s", e)
 
     async def _payment_request(self) -> GetPaymentResponse:
         from .http import _http_origin, post_json

From 9935816429077e0fdcaf107e4f02c0820a8eecb3 Mon Sep 17 00:00:00 2001
From: Rick Staa <rick.staa@outlook.com>
Date: Thu, 30 Jul 2026 12:42:50 +0200
Subject: [PATCH 02/12] feat(live-runner): fund metered single-shot calls while
 they run

A metered single-shot call is billed for as long as it runs, but only
the upfront challenge payment was ever sent, so anything outliving the
signer's preroll (~10s for seconds pricing) was cancelled mid-flight by
the orchestrator. Hold the funding for exactly as long as the work:

- a plain call funds itself while the request is in flight and stops
  the moment the response lands
- a streamed call hands the funding to LiveRunnerCallStream, since the
  stream outlives the call; aclose() stops it, and a released session
  surfaces as stream.released
- payments go to the session-scoped endpoint, built from the challenge
  and the discovered runner id, so a dead session is reported instead
  of silently credited

Fixed pricing is settled upfront and never starts a loop. Offchain
calls have no payment session, so they never start one either.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 src/livepeer_gateway/live_runner.py | 151 +++++++++++++++++++++-------
 1 file changed, 115 insertions(+), 36 deletions(-)

diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py
index fbe3010..9be8562 100644
--- a/src/livepeer_gateway/live_runner.py
+++ b/src/livepeer_gateway/live_runner.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import inspect
 import json
 import logging
@@ -29,7 +30,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_empty, post_json, request_json
 from .remote_signer import (
     GetPaymentResponse,
     LivePaymentSession,
@@ -49,6 +50,9 @@
     "720p-pixel-seconds": "lv2v",
     "fixed": "fixed",
 }
+# Metered types are billed for as long as the work runs, so they need ongoing
+# payments. Fixed pricing is settled by the upfront payment alone.
+_METERED_PAYMENT_TYPES = frozenset({"live", "lv2v"})
 
 # golang format duration, eg "10s"
 _DURATION_RE = re.compile(r"^\s*(?P<value>[0-9]+(?:\.[0-9]+)?)(?P<unit>ns|us|\u00b5s|ms|s|m|h)\s*$")
@@ -129,6 +133,9 @@ class LiveRunnerCallResult:
         repr=False,
         compare=False,
     )
+    # Session-scoped payment endpoint for this call, when one could be built.
+    # Empty means payments fall back to the orchestrator's generic /payment.
+    payment_url: str = ""
 
 
 @dataclass
@@ -145,8 +152,14 @@ class LiveRunnerCallStream:
     runner_url: str
     runner: Optional[LiveRunnerInstance]
     payment_session: Optional[LivePaymentSession]
-    _session: aiohttp.ClientSession = field(repr=False, compare=False)
-    _response: aiohttp.ClientResponse = field(repr=False, compare=False)
+    # True once the orchestrator reported the backing session gone. The stream
+    # itself ends when the orchestrator cancels the proxied request.
+    released: bool = False
+    _session: aiohttp.ClientSession = field(repr=False, compare=False, kw_only=True)
+    _response: aiohttp.ClientResponse = field(repr=False, compare=False, kw_only=True)
+    _payment_task: Optional[asyncio.Task[None]] = field(
+        default=None, repr=False, compare=False, kw_only=True
+    )
 
     @property
     def content_type(self) -> str:
@@ -161,6 +174,9 @@ async def aiter_lines(self) -> AsyncIterator[str]:
             yield line.decode(errors="replace").rstrip("\n")
 
     async def aclose(self) -> None:
+        # Stop funding first: no point minting a payment for a stream we are
+        # about to drop.
+        self._payment_task = await _stop_funding(self._payment_task)
         self._response.release()
         await self._session.close()
 
@@ -296,10 +312,10 @@ async def close(self) -> None:
                 _LOG.warning("Skipping live runner unregister without heartbeat secret")
                 return
             try:
-                await _post_empty(
+                await post_empty(
                     _join_endpoint(self.orchestrator_url, f"/runners/{quote(self.runner_id, safe='')}/unregister"),
-                    {"Authorization": secret},
-                    self._timeout,
+                    headers={"Authorization": secret},
+                    timeout=self._timeout,
                 )
             except Exception:
                 _LOG.debug("Live runner unregister failed", exc_info=True)
@@ -782,6 +798,19 @@ async def call_runner(
             request_headers["Livepeer-Segment"] = payment.seg_creds or ""
             session_id = challenge.manifest_id
 
+        # Metered pricing keeps billing for as long as the work runs, so
+        # whoever holds the session open has to keep paying for it.
+        needs_funding = payment_session is not None and payment_type in _METERED_PAYMENT_TYPES
+        payment_url = (
+            _session_payment_url(
+                challenge.orchestrator_url if challenge is not None else "",
+                runner.runner_id if runner is not None else "",
+                session_id,
+            )
+            if needs_funding
+            else ""
+        )
+
         try:
             request_kwargs: dict[str, Any] = {"timeout": timeout}
             if request_headers:
@@ -796,16 +825,41 @@ async def call_runner(
                     payload=request_payload,
                     headers=request_headers or None,
                 )
-                return LiveRunnerCallStream(
-                    resp.status, resp.headers, runner_url, runner, payment_session, session, resp,
+                call_stream = LiveRunnerCallStream(
+                    status=resp.status,
+                    headers=resp.headers,
+                    runner_url=runner_url,
+                    runner=runner,
+                    payment_session=None if payment_type == "fixed" else payment_session,
+                    _session=session,
+                    _response=resp,
                 )
-
-            data = await request_json(
-                runner_url,
-                method=method,
-                payload=request_payload,
-                **request_kwargs,
+                # The stream outlives this call, so it owns the funding;
+                # aclose() stops both.
+                if needs_funding:
+                    call_stream._payment_task = _start_funding(
+                        cast(LivePaymentSession, payment_session),
+                        payment_url,
+                        lambda: setattr(call_stream, "released", True),
+                    )
+                return call_stream
+
+            # A metered call is billed while we wait on it, so fund it for
+            # exactly as long as the request is in flight.
+            pay_task = (
+                _start_funding(cast(LivePaymentSession, payment_session), payment_url)
+                if needs_funding
+                else None
             )
+            try:
+                data = await request_json(
+                    runner_url,
+                    method=method,
+                    payload=request_payload,
+                    **request_kwargs,
+                )
+            finally:
+                await _stop_funding(pay_task)
             if not isinstance(data, dict):
                 raise LivepeerGatewayError(
                     f"Live runner call expected JSON object, got {type(data).__name__}"
@@ -819,6 +873,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,
+                payment_url=payment_url,
             )
         except LivepeerHTTPError as e:
             if e.status_code != 402:
@@ -863,6 +918,49 @@ def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> _RunnerPaymentC
     )
 
 
+def _session_payment_url(orchestrator_url: str, runner_id: str, session_id: str) -> str:
+    """Build the session-scoped payment endpoint for a live runner session.
+
+    Unlike the orchestrator's generic ``/payment``, which credits the payer
+    balance and returns 200 whether or not the session exists, this endpoint
+    404s once the session is released, which is the only signal a payment loop
+    gets that it is funding nothing. Returns "" when a part is missing, in
+    which case payments fall back to the generic endpoint.
+    """
+    if not (orchestrator_url and runner_id and session_id):
+        return ""
+    try:
+        return _join_endpoint(
+            orchestrator_url,
+            f"/apps/{quote(runner_id, safe='')}/session/{quote(session_id, safe='')}/payment",
+        )
+    except LivepeerGatewayError:
+        return ""
+
+
+def _start_funding(
+    payment_session: LivePaymentSession,
+    payment_url: str,
+    on_released: Optional[Callable[[], None]] = None,
+) -> asyncio.Task[None]:
+    """Run payments in the background for as long as the caller keeps the task."""
+
+    async def _fund() -> None:
+        released = await payment_session.run_payments(payment_url=payment_url or None)
+        if released and on_released is not None:
+            on_released()
+
+    return asyncio.create_task(_fund())
+
+
+async def _stop_funding(task: Optional[asyncio.Task[None]]) -> None:
+    if task is not None and not task.done():
+        task.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await task
+    return None
+
+
 async def _get_runner_payment(
     challenge: _RunnerPaymentChallenge,
     *,
@@ -977,10 +1075,10 @@ async def stop_runner_session(
         url = _join_endpoint(control_url, "stop")
         if isinstance(token, str) and token.strip():
             request_headers = {"Livepeer-Session-Token": token}
-    await _post_empty(
+    await post_empty(
         url,
-        request_headers,
-        timeout,
+        headers=request_headers,
+        timeout=timeout,
     )
 
 
@@ -1146,25 +1244,6 @@ def _is_trickle_channel_response(value: object) -> bool:
     ) and ("internal_url" not in value or isinstance(value.get("internal_url"), str))
 
 
-async def _post_empty(url: str, headers: dict[str, str], timeout: float) -> None:
-    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:
-                body = await resp.text()
-                if resp.status >= 400:
-                    raise LivepeerGatewayError(
-                        f"HTTP empty POST error: HTTP {resp.status}; body={body!r}"
-                    )
-    except LivepeerGatewayError:
-        raise
-    except getattr(aiohttp, "ClientConnectorError", ()) as e:
-        raise LivepeerGatewayError(f"HTTP empty POST error: {getattr(e, 'message', e)}") from e
-    except (aiohttp.ClientError, asyncio.TimeoutError) as e:
-        raise LivepeerGatewayError(f"HTTP empty POST error: {getattr(e, 'message', e)}") from e
-
-
 def _detect_gpu_pynvml() -> Optional[LiveRunnerGPU]:
     try:
         import pynvml  # type: ignore[import-not-found]

From 42bcfe586feb171715e24f48822ad00b74b92d3b Mon Sep 17 00:00:00 2001
From: Rick Staa <rick.staa@outlook.com>
Date: Thu, 30 Jul 2026 12:44:18 +0200
Subject: [PATCH 03/12] feat(live-runner): keep reserved sessions funded for as
 long as they are held

reserve_session paid the reservation challenge and then dropped the
payment session on the floor, so a metered session was released by the
orchestrator seconds later no matter what the client did with it. The
session now owns its funding from the moment it is reserved:

- payments run against the session-scoped endpoint until the session is
  closed; a 404 means the orchestrator let it go, surfaced as
  session.released
- stop_payments() drops funding without releasing the session, for
  draining or handing funding elsewhere; aclose() and the async context
  manager do the same on the way out
- drop _live_runner_session_from_json, which had no callers and built
  sessions that could never fund themselves

Fixed-price and offchain reservations carry no payment session, so they
are unaffected.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 src/livepeer_gateway/live_runner.py | 65 ++++++++++++++++++++---------
 src/livepeer_gateway/selection.py   |  8 +++-
 2 files changed, 52 insertions(+), 21 deletions(-)

diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py
index 9be8562..15c864c 100644
--- a/src/livepeer_gateway/live_runner.py
+++ b/src/livepeer_gateway/live_runner.py
@@ -108,12 +108,56 @@ class LiveRunnerInstance:
     price_info: Optional[LiveRunnerPriceInfo] = None
 
 
-@dataclass(frozen=True)
+@dataclass
 class LiveRunnerSession:
+    """A reserved live runner session.
+
+    A metered session is billed for as long as it is held, so on-chain
+    sessions fund themselves from the moment they are reserved until they are
+    closed. Use it as an async context manager (or call ``aclose()``) to
+    release the session's resources.
+    """
+
     session_id: str
     app_url: str
     runner_url: str
     runner: Optional[LiveRunnerInstance] = None
+    # True once the orchestrator reported this session gone, either because it
+    # was stopped elsewhere or because it ran out of funds.
+    released: bool = False
+    _payment_task: Optional[asyncio.Task[None]] = field(
+        default=None, repr=False, compare=False
+    )
+
+    def _start_payments(
+        self,
+        payment_session: LivePaymentSession,
+        payment_url: str = "",
+    ) -> None:
+        if self._payment_task is not None:
+            return
+        self._payment_task = _start_funding(
+            payment_session,
+            payment_url,
+            lambda: setattr(self, "released", True),
+        )
+
+    async def stop_payments(self) -> None:
+        """Stop funding this session without releasing it.
+
+        Useful to hand funding to something else, or to let a session lapse
+        deliberately. Closing the session stops payments too.
+        """
+        self._payment_task = await _stop_funding(self._payment_task)
+
+    async def aclose(self) -> None:
+        await self.stop_payments()
+
+    async def __aenter__(self) -> LiveRunnerSession:
+        return self
+
+    async def __aexit__(self, *exc: object) -> None:
+        await self.aclose()
 
 
 @dataclass(frozen=True)
@@ -1032,25 +1076,6 @@ def _live_runner_price_info_from_json(value: object) -> Optional[LiveRunnerPrice
     )
 
 
-def _live_runner_session_from_json(
-    data: dict[str, Any],
-    *,
-    runner_url: str,
-    runner: Optional[LiveRunnerInstance],
-) -> LiveRunnerSession:
-    session_id = data.get("session_id")
-    app_url = data.get("app_url")
-    if not isinstance(session_id, str) or not session_id.strip():
-        raise LivepeerGatewayError("Live runner session reserve response missing session_id")
-    if not isinstance(app_url, str) or not app_url.strip():
-        raise LivepeerGatewayError("Live runner session reserve response missing app_url")
-    return LiveRunnerSession(
-        session_id=session_id.strip(),
-        app_url=app_url.strip(),
-        runner_url=runner_url,
-        runner=runner,
-    )
-
 async def stop_runner_session(
     session: LiveRunnerSession | LiveRunnerSessionRequest,
     *,
diff --git a/src/livepeer_gateway/selection.py b/src/livepeer_gateway/selection.py
index 5cb8e82..fbea62b 100644
--- a/src/livepeer_gateway/selection.py
+++ b/src/livepeer_gateway/selection.py
@@ -299,12 +299,18 @@ async def reserve_session(
         raise LivepeerGatewayError("runner session response missing session_id")
     if not isinstance(app_url, str) or not app_url.strip():
         raise LivepeerGatewayError("runner session response missing app_url")
-    return LiveRunnerSession(
+    session = LiveRunnerSession(
         session_id=session_id.strip(),
         app_url=app_url.strip(),
         runner_url=result.runner_url,
         runner=result.runner,
     )
+    # A metered session is billed for as long as it is held, so it funds
+    # itself from here until it is closed. Fixed-price and offchain
+    # reservations have no payment session and need nothing further.
+    if result.payment_session is not None:
+        session._start_payments(result.payment_session, result.payment_url)
+    return session
 
 
 def _runner_candidates_from_discovery(entries: Sequence[dict[str, Any]]) -> list[LiveRunnerInstance]:

From 08a52e6b5e6bf9961410fcdd25e41d7e2ea7bf4c Mon Sep 17 00:00:00 2001
From: Rick Staa <rick.staa@outlook.com>
Date: Thu, 30 Jul 2026 13:14:17 +0200
Subject: [PATCH 04/12] refactor(payments): resolve the payment cadence per
 call

Binding PAYMENT_INTERVAL_S as a default argument freezes it at import,
so overriding the module constant has no effect. Resolve it inside
run_payments instead, which keeps the cadence adjustable in one place.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 src/livepeer_gateway/remote_signer.py | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py
index d21e54c..262d55b 100644
--- a/src/livepeer_gateway/remote_signer.py
+++ b/src/livepeer_gateway/remote_signer.py
@@ -289,7 +289,7 @@ async def run_payments(
         self,
         *,
         payment_url: Optional[str] = None,
-        interval_s: float = PAYMENT_INTERVAL_S,
+        interval_s: Optional[float] = None,
     ) -> bool:
         """Keep a metered session funded until cancelled or the session ends.
 
@@ -302,6 +302,9 @@ async def run_payments(
         previous one, so transient failures are retried rather than fatal:
         the next payment settles the arrears.
         """
+        # Resolved per call rather than as a default argument so the cadence
+        # stays overridable at the module level.
+        interval_s = PAYMENT_INTERVAL_S if interval_s is None else interval_s
         while True:
             await asyncio.sleep(interval_s)
             try:

From df6e880302128e168d8ef90156c4168b367ef6c6 Mon Sep 17 00:00:00 2001
From: Rick Staa <rick.staa@outlook.com>
Date: Thu, 30 Jul 2026 13:25:30 +0200
Subject: [PATCH 05/12] docs(payments): note where the payment cadence should
 come from

livepeer/go-livepeer#4001 adds payment_interval_ms to payment
challenges. Until it lands the client guesses a cadence against the
orchestrator's default debit interval, so record what replaces the
guess.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 src/livepeer_gateway/remote_signer.py | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py
index 262d55b..6d8f95b 100644
--- a/src/livepeer_gateway/remote_signer.py
+++ b/src/livepeer_gateway/remote_signer.py
@@ -26,6 +26,11 @@
 # Client payment cadence. The orchestrator debits metered sessions every
 # -livePaymentInterval (5s by default) and drops the session on the first tick
 # it cannot cover, so pay comfortably ahead of it.
+#
+# TODO: drive this from the orchestrator instead of guessing. Once
+# livepeer/go-livepeer#4001 lands, payment challenges carry
+# payment_interval_ms; read it there and pass it as run_payments(interval_s=),
+# keeping this as the fallback for orchestrators that do not report one.
 PAYMENT_INTERVAL_S = 3.0
 
 @dataclass(frozen=True)

From 4bc2fe52423a989b3d5510d779db1ea36ad61e12 Mon Sep 17 00:00:00 2001
From: Rick Staa <rick.staa@outlook.com>
Date: Thu, 30 Jul 2026 13:34:13 +0200
Subject: [PATCH 06/12] feat(live-runner): take the session payment endpoint
 from control_url

The reservation response reports the session's control URL, which the
orchestrator builds as ServiceURI/apps/{runner}/session/{session}; its
/payment path is the endpoint the funding loop wants. Use it directly
for reserved sessions rather than the endpoint call_runner derives, so
the client follows the orchestrator's own routing instead of
reconstructing it. Single-shot calls keep deriving it, since a payment
challenge carries no control URL.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 src/livepeer_gateway/http.py        |  5 -----
 src/livepeer_gateway/live_runner.py | 10 ++++++++++
 src/livepeer_gateway/selection.py   |  9 ++++++++-
 3 files changed, 18 insertions(+), 6 deletions(-)

diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py
index db4057b..5efe595 100644
--- a/src/livepeer_gateway/http.py
+++ b/src/livepeer_gateway/http.py
@@ -379,11 +379,6 @@ async def post_empty(
 ) -> 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)
diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py
index 15c864c..483c883 100644
--- a/src/livepeer_gateway/live_runner.py
+++ b/src/livepeer_gateway/live_runner.py
@@ -122,6 +122,9 @@ class LiveRunnerSession:
     app_url: str
     runner_url: str
     runner: Optional[LiveRunnerInstance] = None
+    # Base URL for this session's control endpoints, as reported by the
+    # orchestrator when the session was reserved.
+    control_url: str = ""
     # True once the orchestrator reported this session gone, either because it
     # was stopped elsewhere or because it ran out of funds.
     released: bool = False
@@ -129,6 +132,13 @@ class LiveRunnerSession:
         default=None, repr=False, compare=False
     )
 
+    @property
+    def payment_url(self) -> str:
+        """This session's payment endpoint, or "" if it reported no control URL."""
+        if not self.control_url:
+            return ""
+        return _join_endpoint(self.control_url, "payment")
+
     def _start_payments(
         self,
         payment_session: LivePaymentSession,
diff --git a/src/livepeer_gateway/selection.py b/src/livepeer_gateway/selection.py
index fbea62b..be9af88 100644
--- a/src/livepeer_gateway/selection.py
+++ b/src/livepeer_gateway/selection.py
@@ -299,17 +299,24 @@ async def reserve_session(
         raise LivepeerGatewayError("runner session response missing session_id")
     if not isinstance(app_url, str) or not app_url.strip():
         raise LivepeerGatewayError("runner session response missing app_url")
+    control_url = result.data.get("control_url")
     session = LiveRunnerSession(
         session_id=session_id.strip(),
         app_url=app_url.strip(),
         runner_url=result.runner_url,
         runner=result.runner,
+        control_url=control_url.strip() if isinstance(control_url, str) else "",
     )
     # A metered session is billed for as long as it is held, so it funds
     # itself from here until it is closed. Fixed-price and offchain
     # reservations have no payment session and need nothing further.
     if result.payment_session is not None:
-        session._start_payments(result.payment_session, result.payment_url)
+        # The reservation response carries the session's control URL, so
+        # prefer it over the endpoint call_runner had to derive.
+        session._start_payments(
+            result.payment_session,
+            session.payment_url or result.payment_url,
+        )
     return session
 
 

From 2ebce187fede380d7c3406e663a1858dc5706db5 Mon Sep 17 00:00:00 2001
From: Rick Staa <rick.staa@outlook.com>
Date: Thu, 30 Jul 2026 13:40:30 +0200
Subject: [PATCH 07/12] refactor(payments): drop the redundant CancelledError
 re-raise

CancelledError has inherited from BaseException since Python 3.8, and
this package requires 3.12, so the broadest handler in the loop cannot
swallow it. Catching it only to re-raise added nothing.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 src/livepeer_gateway/remote_signer.py | 2 --
 1 file changed, 2 deletions(-)

diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py
index 6d8f95b..52b2682 100644
--- a/src/livepeer_gateway/remote_signer.py
+++ b/src/livepeer_gateway/remote_signer.py
@@ -314,8 +314,6 @@ async def run_payments(
             await asyncio.sleep(interval_s)
             try:
                 await self.send_payment(payment_url=payment_url)
-            except asyncio.CancelledError:
-                raise
             except SkipPaymentCycle as e:
                 _LOG.debug("Payment loop skipped cycle: %s", e)
             except LivepeerHTTPError as e:

From 9ea98d8b53efe513d2a5b742f4612d5e13b845ab Mon Sep 17 00:00:00 2001
From: Rick Staa <rick.staa@outlook.com>
Date: Thu, 30 Jul 2026 14:04:07 +0200
Subject: [PATCH 08/12] docs(payments): explain why run_payments exists and
 what it returns

Say what drives the loop (the orchestrator debits on its own interval
and drops the session on the first tick it cannot cover), what
payment_url buys over the generic endpoint, and what the two return
values mean.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 src/livepeer_gateway/remote_signer.py | 24 ++++++++++++++++--------
 1 file changed, 16 insertions(+), 8 deletions(-)

diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py
index 52b2682..e57be82 100644
--- a/src/livepeer_gateway/remote_signer.py
+++ b/src/livepeer_gateway/remote_signer.py
@@ -298,14 +298,22 @@ async def run_payments(
     ) -> bool:
         """Keep a metered session funded until cancelled or the session ends.
 
-        Returns True when the orchestrator reports the session gone, so the
-        owner can surface it as released; returns False for the other terminal
-        rejections. Cancel the task to stop funding.
-
-        The caller pays upfront before starting this loop, so the first
-        follow-up waits one interval. A payment covers the time since the
-        previous one, so transient failures are retried rather than fatal:
-        the next payment settles the arrears.
+        The orchestrator debits the session's prepaid balance on its own
+        interval and drops the session on the first tick it cannot cover, so
+        whoever holds a session open runs this for as long as they hold it.
+        Cancel the task to stop. The caller pays upfront, so the first payment
+        here waits one ``interval_s`` (default ``PAYMENT_INTERVAL_S``).
+
+        ``payment_url`` should be the session-scoped endpoint, which 404s once
+        the session is gone; without it payments fall back to the
+        orchestrator's generic ``/payment``, which credits the balance whether
+        or not the session still exists.
+
+        Returns True when the session is reported gone, so the owner can
+        surface it as released, and False on the other terminal rejections
+        (fixed price, or credentials naming a different session). Everything
+        else is retried: a payment covers the time since the last one, so the
+        next success settles the arrears.
         """
         # Resolved per call rather than as a default argument so the cadence
         # stays overridable at the module level.

From b486fa465bb06c223bbeba638b77c5fa222e7920 Mon Sep 17 00:00:00 2001
From: Rick Staa <rick.staa@outlook.com>
Date: Thu, 30 Jul 2026 14:11:59 +0200
Subject: [PATCH 09/12] docs(payments): say why the loop stops on any 4xx

The old comment listed only the three known terminal codes, which read
as if they were the only ones that stop the loop. Lead with the rule
instead: a 4xx will not change on a retry, and 408 and 429 are the two
that ask to be retried anyway.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 src/livepeer_gateway/remote_signer.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py
index e57be82..8dce655 100644
--- a/src/livepeer_gateway/remote_signer.py
+++ b/src/livepeer_gateway/remote_signer.py
@@ -325,9 +325,9 @@ async def run_payments(
             except SkipPaymentCycle as e:
                 _LOG.debug("Payment loop skipped cycle: %s", e)
             except LivepeerHTTPError as e:
-                # 404 session released, 409 fixed price, 403 session/payment
-                # mismatch: all terminal, and paying on would mint tickets the
-                # orchestrator will never honor.
+                # A 4xx will not change on a retry (404 gone, 409 fixed price,
+                # 403 mismatch), so stop rather than mint tickets nobody will
+                # honour. 408 and 429 are the two that do ask to be retried.
                 if 400 <= e.status_code < 500 and e.status_code not in (408, 429):
                     _LOG.info("Payment loop stopping (HTTP %d): %s", e.status_code, e)
                     return e.status_code == 404

From 2d03e28770d869878ec5763855a496a8a52ea7dc Mon Sep 17 00:00:00 2001
From: Rick Staa <rick.staa@outlook.com>
Date: Thu, 30 Jul 2026 14:13:22 +0200
Subject: [PATCH 10/12] docs(payments): trim the cadence comment to one line

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 src/livepeer_gateway/remote_signer.py | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py
index 8dce655..1b2dcd1 100644
--- a/src/livepeer_gateway/remote_signer.py
+++ b/src/livepeer_gateway/remote_signer.py
@@ -315,8 +315,7 @@ async def run_payments(
         else is retried: a payment covers the time since the last one, so the
         next success settles the arrears.
         """
-        # Resolved per call rather than as a default argument so the cadence
-        # stays overridable at the module level.
+        # Not a default argument: those bind at import and freeze the constant.
         interval_s = PAYMENT_INTERVAL_S if interval_s is None else interval_s
         while True:
             await asyncio.sleep(interval_s)

From 90a4966269f7275458c5fcf9676ac536dd5876d9 Mon Sep 17 00:00:00 2001
From: Rick Staa <rick.staa@outlook.com>
Date: Thu, 30 Jul 2026 15:52:35 +0200
Subject: [PATCH 11/12] refactor(payments): tighten the payment path after
 review

Behaviour:
- reserve_session no longer falls back to a payment endpoint derived
  from the challenge. An orchestrator reporting no control URL gets
  paid at its generic /payment, where before it would have been sent
  to a route it may not serve and answered with a 404 the loop reads
  as a released session.
- send_payment rejects a payment the signer returned without segCreds
  instead of sending an empty segment header, which fails the
  orchestrator's sig check and comes back 403 - again read as a dead
  session rather than a bad signer.

Shape:
- split the endpoint derivation into _session_control_url and
  _payment_endpoint, so one place knows a payment endpoint sits under
  a session's control URL and the reported and derived URLs agree by
  construction
- drop LiveRunnerCallResult.payment_url, which no longer has a
  consumer, and LiveRunnerSession derives its own from control_url
- give LiveRunnerCallStream the same _start_payments as the session
- drop run_payments' interval_s parameter; the cadence is the module
  constant until the orchestrator reports one
- keep the stream's new fields after the existing ones so its
  constructor call stays positional
- reuse selection.py's own _string_value for the control URL

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 src/livepeer_gateway/live_runner.py   | 125 +++++++++++++-------------
 src/livepeer_gateway/remote_signer.py |  37 +++-----
 src/livepeer_gateway/selection.py     |  15 +---
 3 files changed, 78 insertions(+), 99 deletions(-)

diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py
index 483c883..971c2c0 100644
--- a/src/livepeer_gateway/live_runner.py
+++ b/src/livepeer_gateway/live_runner.py
@@ -122,11 +122,8 @@ class LiveRunnerSession:
     app_url: str
     runner_url: str
     runner: Optional[LiveRunnerInstance] = None
-    # Base URL for this session's control endpoints, as reported by the
-    # orchestrator when the session was reserved.
     control_url: str = ""
-    # True once the orchestrator reported this session gone, either because it
-    # was stopped elsewhere or because it ran out of funds.
+    # True once the orchestrator reported this session gone.
     released: bool = False
     _payment_task: Optional[asyncio.Task[None]] = field(
         default=None, repr=False, compare=False
@@ -134,21 +131,15 @@ class LiveRunnerSession:
 
     @property
     def payment_url(self) -> str:
-        """This session's payment endpoint, or "" if it reported no control URL."""
-        if not self.control_url:
-            return ""
-        return _join_endpoint(self.control_url, "payment")
+        """This session's payment endpoint, or "" when none was reported."""
+        return _payment_endpoint(self.control_url)
 
-    def _start_payments(
-        self,
-        payment_session: LivePaymentSession,
-        payment_url: str = "",
-    ) -> None:
+    def _start_payments(self, payment_session: LivePaymentSession) -> None:
         if self._payment_task is not None:
             return
         self._payment_task = _start_funding(
             payment_session,
-            payment_url,
+            self.payment_url,
             lambda: setattr(self, "released", True),
         )
 
@@ -187,9 +178,6 @@ class LiveRunnerCallResult:
         repr=False,
         compare=False,
     )
-    # Session-scoped payment endpoint for this call, when one could be built.
-    # Empty means payments fall back to the orchestrator's generic /payment.
-    payment_url: str = ""
 
 
 @dataclass
@@ -206,13 +194,12 @@ class LiveRunnerCallStream:
     runner_url: str
     runner: Optional[LiveRunnerInstance]
     payment_session: Optional[LivePaymentSession]
-    # True once the orchestrator reported the backing session gone. The stream
-    # itself ends when the orchestrator cancels the proxied request.
+    _session: aiohttp.ClientSession = field(repr=False, compare=False)
+    _response: aiohttp.ClientResponse = field(repr=False, compare=False)
+    # True once the orchestrator reported the backing session gone.
     released: bool = False
-    _session: aiohttp.ClientSession = field(repr=False, compare=False, kw_only=True)
-    _response: aiohttp.ClientResponse = field(repr=False, compare=False, kw_only=True)
     _payment_task: Optional[asyncio.Task[None]] = field(
-        default=None, repr=False, compare=False, kw_only=True
+        default=None, repr=False, compare=False
     )
 
     @property
@@ -227,9 +214,21 @@ async def aiter_lines(self) -> AsyncIterator[str]:
         async for line in self._response.content:
             yield line.decode(errors="replace").rstrip("\n")
 
+    def _start_payments(
+        self,
+        payment_session: LivePaymentSession,
+        payment_url: str = "",
+    ) -> None:
+        if self._payment_task is not None:
+            return
+        self._payment_task = _start_funding(
+            payment_session,
+            payment_url,
+            lambda: setattr(self, "released", True),
+        )
+
     async def aclose(self) -> None:
-        # Stop funding first: no point minting a payment for a stream we are
-        # about to drop.
+        # Stop funding first: don't pay for a stream we are about to drop.
         self._payment_task = await _stop_funding(self._payment_task)
         self._response.release()
         await self._session.close()
@@ -825,6 +824,8 @@ async def call_runner(
         payment_session: Optional[LivePaymentSession] = None
         payment_type = ""
         session_id = ""
+        needs_ongoing_funding = False
+        payment_url = ""
         request_headers: dict[str, str] = {}
         if signer_url:
             request_headers[_LIVE_RUNNER_PAYER_ADDRESS_HEADER] = payer_address
@@ -852,18 +853,16 @@ async def call_runner(
             request_headers["Livepeer-Segment"] = payment.seg_creds or ""
             session_id = challenge.manifest_id
 
-        # Metered pricing keeps billing for as long as the work runs, so
-        # whoever holds the session open has to keep paying for it.
-        needs_funding = payment_session is not None and payment_type in _METERED_PAYMENT_TYPES
-        payment_url = (
-            _session_payment_url(
-                challenge.orchestrator_url if challenge is not None else "",
-                runner.runner_id if runner is not None else "",
-                session_id,
-            )
-            if needs_funding
-            else ""
-        )
+            # Metered pricing bills for as long as the work runs.
+            needs_ongoing_funding = payment_type in _METERED_PAYMENT_TYPES
+            if needs_ongoing_funding:
+                payment_url = _payment_endpoint(
+                    _session_control_url(
+                        challenge.orchestrator_url,
+                        runner.runner_id if runner is not None else "",
+                        session_id,
+                    )
+                )
 
         try:
             request_kwargs: dict[str, Any] = {"timeout": timeout}
@@ -880,29 +879,25 @@ async def call_runner(
                     headers=request_headers or None,
                 )
                 call_stream = LiveRunnerCallStream(
-                    status=resp.status,
-                    headers=resp.headers,
-                    runner_url=runner_url,
-                    runner=runner,
-                    payment_session=None if payment_type == "fixed" else payment_session,
-                    _session=session,
-                    _response=resp,
+                    resp.status,
+                    resp.headers,
+                    runner_url,
+                    runner,
+                    None if payment_type == "fixed" else payment_session,
+                    session,
+                    resp,
                 )
-                # The stream outlives this call, so it owns the funding;
-                # aclose() stops both.
-                if needs_funding:
-                    call_stream._payment_task = _start_funding(
-                        cast(LivePaymentSession, payment_session),
-                        payment_url,
-                        lambda: setattr(call_stream, "released", True),
+                # The stream outlives this call, so it owns the funding.
+                if needs_ongoing_funding:
+                    call_stream._start_payments(
+                        cast(LivePaymentSession, payment_session), payment_url
                     )
                 return call_stream
 
-            # A metered call is billed while we wait on it, so fund it for
-            # exactly as long as the request is in flight.
+            # The request ends with this call, so the funding ends with it.
             pay_task = (
                 _start_funding(cast(LivePaymentSession, payment_session), payment_url)
-                if needs_funding
+                if needs_ongoing_funding
                 else None
             )
             try:
@@ -927,7 +922,6 @@ 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,
-                payment_url=payment_url,
             )
         except LivepeerHTTPError as e:
             if e.status_code != 402:
@@ -972,26 +966,33 @@ def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> _RunnerPaymentC
     )
 
 
-def _session_payment_url(orchestrator_url: str, runner_id: str, session_id: str) -> str:
-    """Build the session-scoped payment endpoint for a live runner session.
+def _session_control_url(orchestrator_url: str, runner_id: str, session_id: str) -> str:
+    """Rebuild the control URL an orchestrator reports when reserving a session.
 
-    Unlike the orchestrator's generic ``/payment``, which credits the payer
-    balance and returns 200 whether or not the session exists, this endpoint
-    404s once the session is released, which is the only signal a payment loop
-    gets that it is funding nothing. Returns "" when a part is missing, in
-    which case payments fall back to the generic endpoint.
+    Single-shot calls never receive one, since a payment challenge carries only
+    the orchestrator, the runner and the session id. Returns "" if a part is
+    missing.
     """
     if not (orchestrator_url and runner_id and session_id):
         return ""
     try:
         return _join_endpoint(
             orchestrator_url,
-            f"/apps/{quote(runner_id, safe='')}/session/{quote(session_id, safe='')}/payment",
+            f"/apps/{quote(runner_id, safe='')}/session/{quote(session_id, safe='')}",
         )
     except LivepeerGatewayError:
         return ""
 
 
+def _payment_endpoint(control_url: str) -> str:
+    """The session-scoped payment endpoint under a control URL.
+
+    Unlike the generic ``/payment``, it 404s once the session is gone. Returns
+    "" without a control URL, so callers fall back to the generic endpoint.
+    """
+    return _join_endpoint(control_url, "payment") if control_url else ""
+
+
 def _start_funding(
     payment_session: LivePaymentSession,
     payment_url: str,
diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py
index 1b2dcd1..ba379e0 100644
--- a/src/livepeer_gateway/remote_signer.py
+++ b/src/livepeer_gateway/remote_signer.py
@@ -284,41 +284,26 @@ async def send_payment(
             url = f"{_http_origin(target)}/payment"
 
         payment = await self.get_payment()
+        if not payment.seg_creds:
+            # An empty segment header fails the orchestrator's sig check and
+            # comes back 403, which reads as a dead session, not a bad signer.
+            raise PaymentError("Signer returned a payment with no segCreds")
         headers = {
             "Livepeer-Payment": payment.payment,
-            "Livepeer-Segment": payment.seg_creds or "",
+            "Livepeer-Segment": payment.seg_creds,
         }
         await post_empty(url, headers=headers, timeout=5.0)
 
-    async def run_payments(
-        self,
-        *,
-        payment_url: Optional[str] = None,
-        interval_s: Optional[float] = None,
-    ) -> bool:
+    async def run_payments(self, *, payment_url: Optional[str] = None) -> bool:
         """Keep a metered session funded until cancelled or the session ends.
 
-        The orchestrator debits the session's prepaid balance on its own
-        interval and drops the session on the first tick it cannot cover, so
-        whoever holds a session open runs this for as long as they hold it.
-        Cancel the task to stop. The caller pays upfront, so the first payment
-        here waits one ``interval_s`` (default ``PAYMENT_INTERVAL_S``).
-
-        ``payment_url`` should be the session-scoped endpoint, which 404s once
-        the session is gone; without it payments fall back to the
-        orchestrator's generic ``/payment``, which credits the balance whether
-        or not the session still exists.
-
-        Returns True when the session is reported gone, so the owner can
-        surface it as released, and False on the other terminal rejections
-        (fixed price, or credentials naming a different session). Everything
-        else is retried: a payment covers the time since the last one, so the
-        next success settles the arrears.
+        Cancel the task to stop; the first payment waits one interval, since
+        the caller pays upfront. Pass the session-scoped ``payment_url`` to
+        learn when the session is gone, since the generic ``/payment`` credits
+        blindly. Returns True if the orchestrator reported it gone.
         """
-        # Not a default argument: those bind at import and freeze the constant.
-        interval_s = PAYMENT_INTERVAL_S if interval_s is None else interval_s
         while True:
-            await asyncio.sleep(interval_s)
+            await asyncio.sleep(PAYMENT_INTERVAL_S)
             try:
                 await self.send_payment(payment_url=payment_url)
             except SkipPaymentCycle as e:
diff --git a/src/livepeer_gateway/selection.py b/src/livepeer_gateway/selection.py
index be9af88..755f5da 100644
--- a/src/livepeer_gateway/selection.py
+++ b/src/livepeer_gateway/selection.py
@@ -299,24 +299,17 @@ async def reserve_session(
         raise LivepeerGatewayError("runner session response missing session_id")
     if not isinstance(app_url, str) or not app_url.strip():
         raise LivepeerGatewayError("runner session response missing app_url")
-    control_url = result.data.get("control_url")
     session = LiveRunnerSession(
         session_id=session_id.strip(),
         app_url=app_url.strip(),
         runner_url=result.runner_url,
         runner=result.runner,
-        control_url=control_url.strip() if isinstance(control_url, str) else "",
+        control_url=_string_value(result.data.get("control_url")),
     )
-    # A metered session is billed for as long as it is held, so it funds
-    # itself from here until it is closed. Fixed-price and offchain
-    # reservations have no payment session and need nothing further.
+
+    # No payment session means fixed price or offchain: nothing to fund.
     if result.payment_session is not None:
-        # The reservation response carries the session's control URL, so
-        # prefer it over the endpoint call_runner had to derive.
-        session._start_payments(
-            result.payment_session,
-            session.payment_url or result.payment_url,
-        )
+        session._start_payments(result.payment_session)
     return session
 
 

From 6e9d15cf33c1a38c73e35f5e67eaf496a0fc0bbe Mon Sep 17 00:00:00 2001
From: Rick Staa <rick.staa@outlook.com>
Date: Thu, 30 Jul 2026 19:32:50 +0200
Subject: [PATCH 12/12] docs(payments): say what bounds the payment cadence

The signer's opening payment bounds it, not the orchestrator's debit
interval: both sides bill by elapsed time at the same rate. Drops the
pointer to livepeer/go-livepeer#4001, closed for the same reason.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 src/livepeer_gateway/remote_signer.py | 9 +--------
 1 file changed, 1 insertion(+), 8 deletions(-)

diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py
index ba379e0..38e1278 100644
--- a/src/livepeer_gateway/remote_signer.py
+++ b/src/livepeer_gateway/remote_signer.py
@@ -23,14 +23,7 @@
 )
 _LOG = logging.getLogger(__name__)
 
-# Client payment cadence. The orchestrator debits metered sessions every
-# -livePaymentInterval (5s by default) and drops the session on the first tick
-# it cannot cover, so pay comfortably ahead of it.
-#
-# TODO: drive this from the orchestrator instead of guessing. Once
-# livepeer/go-livepeer#4001 lands, payment challenges carry
-# payment_interval_ms; read it there and pass it as run_payments(interval_s=),
-# keeping this as the fallback for orchestrators that do not report one.
+# Must stay under the signer's opening payment: 10s per-second, 60s pixel.
 PAYMENT_INTERVAL_S = 3.0
 
 @dataclass(frozen=True)