Live Runner + Scope support

[j0sh]

Summary by CodeRabbit

• New Features
  • Expanded live runner capabilities: runner discovery, reservation, session proxying, trickle channel setup, and payment-aware calls (including streaming mode).
  • Added media callback support for bytes/frames/packets and live channel event callbacks.
  • Introduced async LRU caching and a dedicated JSON HTTP gateway utility with richer error reporting.
  • Added new echo (blur/runtime updates), ping/pong WebSocket, and text streaming demos.
• Documentation
  • Added README and run instructions for the new demos, plus sample runner configuration.
• Refactor
  • Updated scope startup to use live runners and raised minimum Python requirement to 3.12.

[rickstaa]

@j0sh can we use something like unit_scale since this runner also allows different pricing schemes. Also see my pull request in go-liveper livepeer/go-livepeer#3942.

[j0sh]

I'll look into that idea (and the general move from away from pixels to purely timing) but don't want to block this merge for that

[rickstaa]

@j0sh yea no blocker from my side I see this as a nice follow up improvement for consistency. I am tracking this in #27.

[rickstaa]

@j0sh How do you intend pricing to work for dynamic runners? Right now the app self-asserts price_info via register_runner(price_per_unit=…) and go-livepeer trusts it (only > 0 is checked in normalizeHeartbeat). That's fine for operator-deployed/trusted containers — the operator sets the price via env and the app forwards it — but an untrusted image could ignore that and under-report its price. Static sidesteps this (operator sets price_info in runners.json); only dynamic trusts the app.

You can see how I'm currently using this in the hello_world example, which feels a bit strange since it relies on the app to create the argument and forward it.

Two thoughts:

• If the orchestrator is still meant to set the price in dynamic mode (to make gaming harder), the SDK could auto-read it from env (e.g. PRICE_PER_UNIT) instead of the app passing it to register_runner, keeping pricing a deployment/operator concern, out of app code.
• I think we'll eventually move to GPU-based pricing (orchestrator states a price per GPU type, workloads auto-run at that rate), so go-livepeer would override the reported price at registration anyway.

[j0sh]

That's correct, the runner itself reports the price because the orchestrator is intended to control the runner.

[j0sh]

This way runner provisioning (including pricing) can be configured separately from go-livepeer without having to introduce a mutual dependency on one other. Setting the price on go-livepeer itself introduces a tension with the orchestrator needing to be configured separately with details of the runner / workload, hardware, etc, as opposed to runners being able to simply connect and go.

If you want to keep the configuration / pricing within go-livepeer then static configuration is the way to go.

[coderabbitai]

Actionable comments posted: 15

🧹 Nitpick comments (13)

src/livepeer_gateway/lv2v.py (1)

124-145: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

on_bytes is not exposed here.

MediaOutput accepts on_bytes/on_frame/on_packet, but this helper forwards only the latter two, so the byte-stream callback is unreachable through LiveVideoToVideo.media_output(...). Add it for parity unless the omission is deliberate.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/livepeer_gateway/lv2v.py` around lines 124 - 145, Expose the on_bytes
callback in LiveVideoToVideo.media_output by adding it to the helper’s
parameters and forwarding it to the MediaOutput constructor alongside on_frame
and on_packet. Preserve all existing callback behavior and defaults.

src/livepeer_gateway/media_output.py (1)

198-254: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

start_callbacks() in __init__ binds tasks to whatever loop is current.

Reasonable given the documented fallback, but note the constructor now has an implicit side effect (task creation) whenever any callback is supplied; MediaOutput(...) built in one loop and used in another will fail late. Worth calling out in the class docstring alongside the existing Attributes: list.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/livepeer_gateway/media_output.py` around lines 198 - 254, Update the
MediaOutput class docstring to document that providing callbacks causes __init__
to invoke start_callbacks() and bind callback tasks to the currently running
event loop. Mention that constructing the instance in one loop and using it in
another is unsupported or may fail, while preserving the existing
no-running-loop fallback behavior.

src/livepeer_gateway/channel_reader.py (1)

213-242: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Both subclass __init__ bodies are identical; consider moving the constructor onto _ChannelReaderCallback.

ChannelReader.__init__ and JSONLReader.__init__ differ only in docstring wording. Defining __init__ on the mixin (and dropping _init_callback) removes the duplication and one indirection.

Also applies to: 337-366

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/livepeer_gateway/channel_reader.py` around lines 213 - 242, Move the
shared constructor implementation from ChannelReader and JSONLReader onto
_ChannelReaderCallback, preserving the existing parameters, defaults, and
callback initialization behavior. Remove the _init_callback indirection and the
duplicated subclass __init__ methods, leaving both readers to inherit the mixin
constructor and retain appropriate documentation for the shared behavior.

src/livepeer_gateway/async_cache.py (1)

19-29: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick win

Cache misses on falsy/None results and no in-flight de-duplication.

Two small gaps worth considering:

• cached is not None conflates "absent" with "cached None". Use a sentinel so None/falsy results are cached correctly.
• Concurrent callers awaiting the same key all execute func, so get_signer_info can fire N simultaneous signer requests before the first result lands. Caching the awaitable (or guarding with a per-key lock) makes it a true single-flight cache.

♻️ Sentinel + single-flight sketch

+        _MISSING = object()
+        inflight: dict[Any, asyncio.Future[_T]] = {}
+
         `@wraps`(func)
         async def wrapper(*args: Any, **kwargs: Any) -> _T:
             key = (args, tuple(sorted(kwargs.items())))
-            cached = cache.get(key)
-            if cached is not None:
+            cached = cache.get(key, _MISSING)
+            if cached is not _MISSING:
                 cache.move_to_end(key)
-                return cached
-
-            value = await func(*args, **kwargs)
+                return cast(_T, cached)
+
+            existing = inflight.get(key)
+            if existing is not None:
+                return await asyncio.shield(existing)
+            fut: asyncio.Future[_T] = asyncio.get_running_loop().create_future()
+            inflight[key] = fut
+            try:
+                value = await func(*args, **kwargs)
+            except BaseException as e:
+                fut.set_exception(e)
+                raise
+            finally:
+                inflight.pop(key, None)
+            if not fut.done():
+                fut.set_result(value)
             cache[key] = value

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/livepeer_gateway/async_cache.py` around lines 19 - 29, Update the async
cache wrapper around cache.get and func(*args, **kwargs) to use a unique
sentinel for detecting absent keys, preserving cached None and other falsy
results. Add per-key in-flight de-duplication so concurrent misses await the
same computation instead of invoking func repeatedly, while retaining the
existing LRU insertion and eviction behavior.

src/livepeer_gateway/live_runner.py (2)

512-513: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick win

Blocking GPU detection runs on the event loop.

detect_process_gpu() shells out to nvidia-smi twice (2s timeout each) and initializes NVML synchronously inside an async def. That can stall the loop for several seconds during startup. Offloading keeps register_runner cooperative.

♻️ Proposed change

     if gpu is None and auto_detect_gpu:
-        gpu = detect_process_gpu()
+        gpu = await asyncio.to_thread(detect_process_gpu)

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/livepeer_gateway/live_runner.py` around lines 512 - 513, Update the GPU
auto-detection branch in register_runner to run detect_process_gpu in a worker
thread or equivalent executor instead of directly on the event loop. Preserve
the existing condition and assign the detection result to gpu after awaiting the
offloaded call.

---

1145-1154: 🔒 Security & Privacy | 🔵 Trivial | 💤 Low value

Static-analysis "command injection" hits on nvidia-smi are false positives.

Both subprocess.check_output calls use a fixed argv list with no interpolated input, so CWE-78 does not apply. The only substantive part of the hint is S607 (partial executable path); since shutil.which("nvidia-smi") is already resolved at line 1126, passing that resolved path through would silence it and avoid PATH surprises.

Also applies to: 1167-1176

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/livepeer_gateway/live_runner.py` around lines 1145 - 1154, Update both
subprocess.check_output calls in the live runner to invoke the executable path
already resolved by shutil.which("nvidia-smi"), rather than the literal
"nvidia-smi" string. Keep the fixed argument lists and existing command behavior
unchanged.

Source: Linters/SAST tools

src/livepeer_gateway/remote_signer.py (1)

271-280: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Duplicated, redundant aiohttp.ClientConnectorError handler. ClientConnectorError subclasses ClientError, so in both places the extra arm is unreachable-by-necessity and builds an identical error; the getattr(..., ()) fallback would also silently catch nothing if the attribute ever moved.

• src/livepeer_gateway/remote_signer.py#L271-L280: drop the getattr(aiohttp, "ClientConnectorError", ()) arm in send_payment and keep only (aiohttp.ClientError, asyncio.TimeoutError).
• src/livepeer_gateway/live_runner.py#L1054-L1059: drop the same arm in _post_empty and keep only (aiohttp.ClientError, asyncio.TimeoutError).

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/livepeer_gateway/remote_signer.py` around lines 271 - 280, Remove the
redundant ClientConnectorError handlers from send_payment in
src/livepeer_gateway/remote_signer.py lines 271-280 and _post_empty in
src/livepeer_gateway/live_runner.py lines 1054-1059; retain only the shared
aiohttp.ClientError and asyncio.TimeoutError handler in both locations,
preserving the existing PaymentError conversion and message.

src/livepeer_gateway/scope.py (1)

82-91: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Tighten the helper signatures.

_start_scope_with_runner has no return annotation (it returns LiveVideoToVideo), and _is_serverless_runner takes object + getattr even though the only caller passes Optional[LiveRunnerInstance]. Typing both concretely lets mypy check the raw access.

♻️ Proposed change

     orch_url: Optional[Sequence[str] | str],
     timeout: float,
-):
+) -> LiveVideoToVideo:

-def _is_serverless_runner(runner: object) -> bool:
-    raw = getattr(runner, "raw", None)
-    version = raw.get("version") if isinstance(raw, dict) else None
+def _is_serverless_runner(runner: Optional[LiveRunnerInstance]) -> bool:
+    version = runner.raw.get("version") if runner is not None else None
     return isinstance(version, str) and version.startswith("serverless")

Also applies to: 140-143

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/livepeer_gateway/scope.py` around lines 82 - 91, Update
_start_scope_with_runner to annotate its return type as LiveVideoToVideo, and
tighten _is_serverless_runner to accept Optional[LiveRunnerInstance] instead of
object. Replace the generic getattr-based raw access with typed access so mypy
validates the caller’s LiveRunnerInstance handling.

src/livepeer_gateway/selection.py (1)

295-301: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win

Reserved runner sessions are never released when post-selection handling fails. call_runner may have already reserved (and paid for) capacity by the time the caller validates or post-processes the response; both sites abandon that reservation on failure, leaving the runner holding capacity until its own timeout.

• src/livepeer_gateway/selection.py#L295-L301: before raising on missing session_id/app_url, issue a best-effort stop_runner_session for the reserved session.
• src/livepeer_gateway/scope.py#L103-L137: in the except Exception arm, best-effort release the reserved session on result before appending the RunnerRejection and advancing to the next candidate — otherwise each failed candidate leaks one reservation.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/livepeer_gateway/selection.py` around lines 295 - 301, Release reserved
runner sessions when post-selection handling fails: in
src/livepeer_gateway/selection.py lines 295-301, best-effort call
stop_runner_session for the reserved session before raising for missing
session_id or app_url; in src/livepeer_gateway/scope.py lines 103-137, update
the except Exception path to best-effort release the session from result before
recording RunnerRejection and advancing to the next candidate.

src/livepeer_gateway/http.py (2)

282-282: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Unnecessary getattr fallback for aiohttp.ClientConnectorError.

aiohttp.ClientConnectorError has existed since long before the pinned aiohttp 3.9.0; getattr(aiohttp, "ClientConnectorError", ()) adds indirection without a real compatibility benefit.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/livepeer_gateway/http.py` at line 282, Update the exception handler
around the aiohttp request flow to directly reference
aiohttp.ClientConnectorError instead of using getattr with an empty-tuple
fallback. Preserve the existing handling and exception alias.

---

142-163: 🔒 Security & Privacy | 🔵 Trivial | ⚡ Quick win

Shared transport functions don't enforce http/https scheme validation.

_parse_http_url (defined later in this same file) restricts URLs to http/https, but request_json_sync and request_json accept raw url strings and pass them straight to Request(...)/session.request(...) without going through it. Current callers happen to validate upstream, but centralizing the check here would close off file:///other-scheme risks (flagged by static analysis as S310/urlopen-unsanitized-data) for any future caller that forgets to validate.

🛡️ Proposed fix

 def request_json_sync(
     url: str,
     *,
     method: Optional[str] = None,
     payload: Optional[dict[str, Any]] = None,
     headers: Optional[dict[str, str]] = None,
     timeout: float = 5.0,
 ) -> Any:
+    url = _parse_http_url(url).geturl()
     resolved_method, req_headers, body = _json_request_parts(

(apply the equivalent change to request_json)

Also applies to: 243-269

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/livepeer_gateway/http.py` around lines 142 - 163, Validate the URL
through the existing _parse_http_url helper at the start of both
request_json_sync and request_json before constructing Request or calling
session.request. Use the validated result for the request while preserving the
existing request method, headers, payload, timeout, and error behavior.

src/livepeer_gateway/orchestrator.py (1)

23-26: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Sync/async naming collision between orchestrator.py and http.py.

orchestrator.post_json/get_json/request_json are synchronous (aliases of the _sync helpers), while http.py defines coroutines with the exact same names. Importing the wrong module in async code (e.g. from .orchestrator import post_json instead of .http) would silently block the event loop instead of raising an obvious error. The existing comment documents intent but doesn't prevent the mix-up.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/livepeer_gateway/orchestrator.py` around lines 23 - 26, Remove or rename
the compatibility aliases request_json, post_json, and get_json in
orchestrator.py so synchronous helpers cannot collide with the asynchronous
functions in http.py; update any callers to use the explicit *_sync names while
preserving synchronous behavior.

src/livepeer_gateway/errors.py (1)

42-55: 📐 Maintainability & Code Quality | 🔵 Trivial

Inconsistent __str__ formatting across rejection-based errors.

NoRunnerAvailableError now formats rejections into its str() output, but the existing sibling error at lines 34-39 (constructed the same way with OrchestratorRejection rejections) does not override __str__ similarly. Operators debugging orchestrator-selection failures get less detail than runner-selection failures despite an identical mechanism.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/livepeer_gateway/errors.py` around lines 42 - 55, The sibling
rejection-based error for orchestrator selection should format its stored
OrchestratorRejection entries in __str__, matching NoRunnerAvailableError.
Update that error class to return the base message when there are no rejections
and otherwise append each rejection’s URL and reason in the same format.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@examples/echo/client.py`:
- Around line 143-152: Update the MediaOutput context flow around _publish_video
so it awaits output stream completion or all pending _write_chunk callback tasks
before exiting the context. Keep the existing publish-complete log, and ensure
MediaOutput is not closed until transformed frames have been fully consumed.
- Around line 85-91: Update both post_json calls for the initial /echo request
and subsequent /update blur requests in the client flow to pass
session.session_id through the supported request-header argument as
Livepeer-Session-Id, ensuring both endpoints receive the reserved session ID.

In `@examples/echo/README.md`:
- Around line 18-20: Update the example command in the README to pass the
configured non-TLS orchestrator URL, changing the runner’s --orchestrator value
from https://localhost:8935 to http://localhost:8935. Keep the existing command
and other arguments unchanged.

In `@examples/echo/runner.py`:
- Around line 191-201: Retain the result of register_runner in _on_startup by
storing registration on the app, then update _on_cleanup to await that
registration’s close() after the pipeline is closed. Ensure cleanup uses the
stored registration so its heartbeat stops and the runner is unregistered.
- Around line 132-168: Serialize singleton session lifecycle access with one
app-scoped asyncio.Lock: use it to guard _handle_echo across the existing state
check and pipeline creation/assignment, and also guard _handle_update,
_close_pipeline, and cleanup reads or writes of state. Ensure concurrent
requests cannot create duplicate MediaOutput/MediaPublish pipelines and that
cleanup safely coordinates with session access.

In `@examples/ping-pong/client.py`:
- Around line 29-33: Update _select_runner to invoke cursor.candidates() before
iteration, so the loop consumes the returned candidate collection rather than
the bound method. Preserve the existing URL return and LivepeerGatewayError
behavior.

In `@examples/text/go-livepeer.conf`:
- Around line 4-5: Update the httpAddr and serviceAddr configuration values to
use host:port format without the http:// scheme, such as localhost:8935, while
preserving the existing port.

In `@examples/text/README.md`:
- Line 3: Correct the platform name from “Livpeeer” to “Livepeer” in the README
description, and add an appropriate shell language identifier to the fenced
command block on line 37 without changing its command content.

In `@src/livepeer_gateway/channel_reader.py`:
- Around line 114-121: Update wait_callback() to exclude asyncio.TimeoutError
from _record_callback_error while preserving propagation of the timeout and
recording other callback exceptions. Keep close() behavior unchanged so
graceful-wait timeouts do not become recorded errors that are later re-raised.

In `@src/livepeer_gateway/http.py`:
- Around line 142-169: Make TLS certificate and hostname verification explicit
and configurable in both request_json_sync and request_json instead of
unconditionally disabling it. Add or reuse a clear verification option, pass it
through the sync SSL context and the async aiohttp.TCPConnector configuration,
and preserve secure verification as the default while allowing self-signed
certificates only when explicitly enabled.

In `@src/livepeer_gateway/live_runner.py`:
- Around line 1043-1048: Update _post_empty to preserve TLS certificate
verification by default: remove the unconditional
aiohttp.TCPConnector(ssl=False) behavior and either use the shared HTTP helper
or add an explicit verify_tls option defaulting to True, only disabling
verification when callers opt in. Ensure calls transmitting Authorization or
Livepeer-Session-Token remain verified.

In `@src/livepeer_gateway/media_output.py`:
- Around line 256-272: The wait_callback helpers must use non-destructive
bounded waits. In src/livepeer_gateway/media_output.py lines 256-272, update
wait_callbacks to use asyncio.wait(callback_tasks, timeout=timeout), collect
results only from completed tasks, and preserve callback error propagation; in
src/livepeer_gateway/channel_reader.py lines 158-184, update wait_callback to
wait on the task without cancelling it when the timeout expires, and document
the intentional asyncio.TimeoutError no-op so close() can cancel it afterward.
- Around line 549-567: Update MediaOutput.close around wait_callbacks so
callback-consumer errors are collected rather than allowed to abort cleanup;
handle the failure from wait_callbacks alongside timeout handling, then continue
cancelling remaining callback tasks, closing all segments, and closing self._sub
before propagating the collected callback error.

In `@src/livepeer_gateway/remote_signer.py`:
- Around line 255-258: Update the headers construction in _payment_request so
Livepeer-Segment always receives a string when payment.seg_creds is None,
matching the existing PaymentSession.send_payment behavior by using an
empty-string fallback.

In `@src/livepeer_gateway/selection.py`:
- Around line 242-256: Update the selection logic around
discover_orchestrator_runners and discover_runners to normalize orchestrators
through the existing orchestrator_discovery_urls helper, treating None, empty
collections, and whitespace-only values as absent. Only use the orchestrator
path when normalization yields URLs; otherwise fall back to discover_runners
with the configured discovery_url and signer_url.

---

Nitpick comments:
In `@src/livepeer_gateway/async_cache.py`:
- Around line 19-29: Update the async cache wrapper around cache.get and
func(*args, **kwargs) to use a unique sentinel for detecting absent keys,
preserving cached None and other falsy results. Add per-key in-flight
de-duplication so concurrent misses await the same computation instead of
invoking func repeatedly, while retaining the existing LRU insertion and
eviction behavior.

In `@src/livepeer_gateway/channel_reader.py`:
- Around line 213-242: Move the shared constructor implementation from
ChannelReader and JSONLReader onto _ChannelReaderCallback, preserving the
existing parameters, defaults, and callback initialization behavior. Remove the
_init_callback indirection and the duplicated subclass __init__ methods, leaving
both readers to inherit the mixin constructor and retain appropriate
documentation for the shared behavior.

In `@src/livepeer_gateway/errors.py`:
- Around line 42-55: The sibling rejection-based error for orchestrator
selection should format its stored OrchestratorRejection entries in __str__,
matching NoRunnerAvailableError. Update that error class to return the base
message when there are no rejections and otherwise append each rejection’s URL
and reason in the same format.

In `@src/livepeer_gateway/http.py`:
- Line 282: Update the exception handler around the aiohttp request flow to
directly reference aiohttp.ClientConnectorError instead of using getattr with an
empty-tuple fallback. Preserve the existing handling and exception alias.
- Around line 142-163: Validate the URL through the existing _parse_http_url
helper at the start of both request_json_sync and request_json before
constructing Request or calling session.request. Use the validated result for
the request while preserving the existing request method, headers, payload,
timeout, and error behavior.

In `@src/livepeer_gateway/live_runner.py`:
- Around line 512-513: Update the GPU auto-detection branch in register_runner
to run detect_process_gpu in a worker thread or equivalent executor instead of
directly on the event loop. Preserve the existing condition and assign the
detection result to gpu after awaiting the offloaded call.
- Around line 1145-1154: Update both subprocess.check_output calls in the live
runner to invoke the executable path already resolved by
shutil.which("nvidia-smi"), rather than the literal "nvidia-smi" string. Keep
the fixed argument lists and existing command behavior unchanged.

In `@src/livepeer_gateway/lv2v.py`:
- Around line 124-145: Expose the on_bytes callback in
LiveVideoToVideo.media_output by adding it to the helper’s parameters and
forwarding it to the MediaOutput constructor alongside on_frame and on_packet.
Preserve all existing callback behavior and defaults.

In `@src/livepeer_gateway/media_output.py`:
- Around line 198-254: Update the MediaOutput class docstring to document that
providing callbacks causes __init__ to invoke start_callbacks() and bind
callback tasks to the currently running event loop. Mention that constructing
the instance in one loop and using it in another is unsupported or may fail,
while preserving the existing no-running-loop fallback behavior.

In `@src/livepeer_gateway/orchestrator.py`:
- Around line 23-26: Remove or rename the compatibility aliases request_json,
post_json, and get_json in orchestrator.py so synchronous helpers cannot collide
with the asynchronous functions in http.py; update any callers to use the
explicit *_sync names while preserving synchronous behavior.

In `@src/livepeer_gateway/remote_signer.py`:
- Around line 271-280: Remove the redundant ClientConnectorError handlers from
send_payment in src/livepeer_gateway/remote_signer.py lines 271-280 and
_post_empty in src/livepeer_gateway/live_runner.py lines 1054-1059; retain only
the shared aiohttp.ClientError and asyncio.TimeoutError handler in both
locations, preserving the existing PaymentError conversion and message.

In `@src/livepeer_gateway/scope.py`:
- Around line 82-91: Update _start_scope_with_runner to annotate its return type
as LiveVideoToVideo, and tighten _is_serverless_runner to accept
Optional[LiveRunnerInstance] instead of object. Replace the generic
getattr-based raw access with typed access so mypy validates the caller’s
LiveRunnerInstance handling.

In `@src/livepeer_gateway/selection.py`:
- Around line 295-301: Release reserved runner sessions when post-selection
handling fails: in src/livepeer_gateway/selection.py lines 295-301, best-effort
call stop_runner_session for the reserved session before raising for missing
session_id or app_url; in src/livepeer_gateway/scope.py lines 103-137, update
the except Exception path to best-effort release the session from result before
recording RunnerRejection and advancing to the next candidate.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: 79912215-a131-4a38-9289-5ea391d7d1f1

📥 Commits

Reviewing files that changed from the base of the PR and between cc31b56 and 89ac0f7.

⛔ Files ignored due to path filters (1)

• uv.lock is excluded by !**/*.lock

📒 Files selected for processing (27)

• examples/echo/README.md
• examples/echo/client.py
• examples/echo/runner.py
• examples/get_orchestrator_info.py
• examples/ping-pong/README.md
• examples/ping-pong/client.py
• examples/ping-pong/runner.py
• examples/text/README.md
• examples/text/go-livepeer.conf
• examples/text/runner.py
• examples/text/runners.json
• examples/text/story.txt
• pyproject.toml
• src/livepeer_gateway/__init__.py
• src/livepeer_gateway/async_cache.py
• src/livepeer_gateway/channel_reader.py
• src/livepeer_gateway/discovery.py
• src/livepeer_gateway/errors.py
• src/livepeer_gateway/http.py
• src/livepeer_gateway/live_runner.py
• src/livepeer_gateway/lv2v.py
• src/livepeer_gateway/media_output.py
• src/livepeer_gateway/orch_info.py
• src/livepeer_gateway/orchestrator.py
• src/livepeer_gateway/remote_signer.py
• src/livepeer_gateway/scope.py
• src/livepeer_gateway/selection.py

[coderabbitai]

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

Send the reserved session ID to the runner.

examples/echo/runner.py requires Livepeer-Session-Id on both endpoints, but these post_json calls omit it. The initial /echo call and blur updates will receive HTTP 400. Pass session.session_id through the supported request-header argument on both calls.

Also applies to: 131-149

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@examples/echo/client.py` around lines 85 - 91, Update both post_json calls
for the initial /echo request and subsequent /update blur requests in the client
flow to pass session.session_id through the supported request-header argument as
Livepeer-Session-Id, ensuring both endpoints receive the reserved session ID.

[coderabbitai]

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

Wait for output consumption before closing it.

Line 151 claims to drain output, but the context exits immediately afterward and closes MediaOutput; transformed frames still in transit can be truncated. Await stream completion or its callback tasks after _publish_video() and before leaving the context.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@examples/echo/client.py` around lines 143 - 152, Update the MediaOutput
context flow around _publish_video so it awaits output stream completion or all
pending _write_chunk callback tasks before exiting the context. Keep the
existing publish-complete log, and ensure MediaOutput is not closed until
transformed frames have been fully consumed.

[coderabbitai]

🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win

Use the configured HTTP orchestrator URL.

The command starts go-livepeer without TLS but invokes the runner with https://localhost:8935; registration cannot connect. Change this to http://localhost:8935 unless TLS setup instructions are added.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@examples/echo/README.md` around lines 18 - 20, Update the example command in
the README to pass the configured non-TLS orchestrator URL, changing the
runner’s --orchestrator value from https://localhost:8935 to
http://localhost:8935. Keep the existing command and other arguments unchanged.

[coderabbitai]

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== files =="
git ls-files | rg '(^|/)runner\.py$|echo' || true

echo "== target outline =="
ast-grep outline examples/echo/runner.py --view expanded || true

echo "== target lines 1-240 =="
cat -n examples/echo/runner.py | sed -n '1,240p'

echo "== usages of state/pipeline/close/lock =="
rg -n "state|_close_pipeline|create_trickle_channels|asyncio\.Lock|add_done_callback|EchoSession" examples/echo/runner.py -C 3

Repository: livepeer/livepeer-python-gateway

Length of output: 11994

---

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== asyncio concurrency interleaving probe =="
python3 - <<'PY'
import asyncio

state = None
creations = []
close_runs = []

async def create():
    global state
    await asyncio.sleep(0.01)  # "suspends after single-session check"
    creation = hash(state) % 1000 if state is not None else 999
    creations.append(creation)
    state = "pipeline"
    print("created", creation, "state=", id(state))

async def close_pipeline():
    global state
    if state is None:
        return
    current = state
    state = None
    close_runs.append(current)
    await asyncio.sleep(0.01)

async def callback():
    asyncio.create_task(close_pipeline())

async def handle():
    global state
    if state is not None:
        return "conflict"
    await create()
    # simulate immediate done callback scheduling
    await callback()
    return "ok"

async def main():
    tasks = [handle(), handle()]
    results = await asyncio.gather(*tasks)
    print("results", results)
    print("creations", creations)
    await asyncio.sleep(0.02)
    print("final state", state, "close runs", close_runs)

asyncio.run(main())
PY

echo "== current handlers acquire any lock? =="
rg -n "lock|asyncio|Lock|lock_obj|_lock" .

Repository: livepeer/livepeer-python-gateway

Length of output: 32303

---

Serialize singleton session creation.

state has an awaitable check/create gap (_handle_echo awaits at create_trickle_channels/request.read() before assigning state), so concurrent requests with the same session can both create MediaOutput/MediaPublish pipelines and only one is closed. Keep one app-scoped asyncio.Lock and guard _handle_echo, _handle_update, _close_pipeline, and cleanup access.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@examples/echo/runner.py` around lines 132 - 168, Serialize singleton session
lifecycle access with one app-scoped asyncio.Lock: use it to guard _handle_echo
across the existing state check and pipeline creation/assignment, and also guard
_handle_update, _close_pipeline, and cleanup reads or writes of state. Ensure
concurrent requests cannot create duplicate MediaOutput/MediaPublish pipelines
and that cleanup safely coordinates with session access.

[coderabbitai]

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

Close the runner registration during app cleanup.

registration is discarded after startup, so its heartbeat task is never closed and the runner is not unregistered. Store it on app and call await registration.close() from _on_cleanup after closing the pipeline.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@examples/echo/runner.py` around lines 191 - 201, Retain the result of
register_runner in _on_startup by storing registration on the app, then update
_on_cleanup to await that registration’s close() after the pipeline is closed.
Ensure cleanup uses the stored registration so its heartbeat stops and the
runner is unregistered.

[coderabbitai]

🔒 Security & Privacy | 🟠 Major | ⚡ Quick win

TLS verification is disabled for every _post_empty call, including ones carrying secrets.

aiohttp.TCPConnector(ssl=False) turns off certificate verification unconditionally. This path sends the runner heartbeat secret (Authorization header in close(), line 246-250) and the session token (Livepeer-Session-Token in stop_runner_session, line 872-878) to an https:// orchestrator, so a MITM can intercept both. Nothing else in this PR (http.post_json, LivePaymentSession.send_payment) disables verification, so this is an outlier rather than a project-wide convention.

Make it opt-in (e.g. plumb a verify_tls: bool = True / reuse the shared http helpers) rather than the default.

🔒 Proposed change

-async def _post_empty(url: str, headers: dict[str, str], timeout: float) -> None:
+async def _post_empty(
+    url: str,
+    headers: dict[str, str],
+    timeout: float,
+    *,
+    verify_tls: bool = True,
+) -> None:
     try:
         client_timeout = aiohttp.ClientTimeout(total=timeout)
-        connector = aiohttp.TCPConnector(ssl=False)
+        connector = aiohttp.TCPConnector() if verify_tls else aiohttp.TCPConnector(ssl=False)

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	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:
	async def _post_empty(
	url: str,
	headers: dict[str, str],
	timeout: float,
	*,
	verify_tls: bool = True,
	) -> None:
	try:
	client_timeout = aiohttp.ClientTimeout(total=timeout)
	connector = aiohttp.TCPConnector() if verify_tls else 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:

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/livepeer_gateway/live_runner.py` around lines 1043 - 1048, Update
_post_empty to preserve TLS certificate verification by default: remove the
unconditional aiohttp.TCPConnector(ssl=False) behavior and either use the shared
HTTP helper or add an explicit verify_tls option defaulting to True, only
disabling verification when callers opt in. Ensure calls transmitting
Authorization or Livepeer-Session-Token remain verified.

[coderabbitai]

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

Timeout waits cancel the callback consumers in both readers. Both wait helpers use asyncio.wait_for, which cancels the awaited task/gather when the timeout expires, so a bounded "wait for completion" permanently stops the consumer instead of leaving it running.

• src/livepeer_gateway/media_output.py#L256-L272: replace asyncio.wait_for(asyncio.gather(...)) with asyncio.wait(callback_tasks, timeout=timeout) and collect results from the done set.
• src/livepeer_gateway/channel_reader.py#L158-L184: make wait_callback() non-destructive (e.g. asyncio.wait({task}, timeout=timeout)) so close()'s graceful phase really precedes cancellation, and document the except asyncio.TimeoutError no-op.

📍 Affects 2 files

• src/livepeer_gateway/media_output.py#L256-L272 (this comment)
• src/livepeer_gateway/channel_reader.py#L158-L184

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/livepeer_gateway/media_output.py` around lines 256 - 272, The
wait_callback helpers must use non-destructive bounded waits. In
src/livepeer_gateway/media_output.py lines 256-272, update wait_callbacks to use
asyncio.wait(callback_tasks, timeout=timeout), collect results only from
completed tasks, and preserve callback error propagation; in
src/livepeer_gateway/channel_reader.py lines 158-184, update wait_callback to
wait on the task without cancelling it when the timeout expires, and document
the intentional asyncio.TimeoutError no-op so close() can cancel it afterward.

[coderabbitai]

🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win

close() leaks the subscriber and segments when a callback failed.

wait_callbacks() re-raises self._callback_errors[0], and only asyncio.TimeoutError is caught here. So whenever a callback consumer failed, close() propagates at Line 554 and never cancels the remaining tasks, closes the segments, or closes self._sub — the aiohttp session stays open. Collect the error instead of letting it escape early.

🐛 Proposed fix

             if wait_callbacks and (timeout is None or timeout > 0):
                 try:
                     await self.wait_callbacks(timeout=timeout)
-                except asyncio.TimeoutError:
-                    pass
+                except asyncio.TimeoutError:
+                    _LOG.debug("MediaOutput callback wait timed out; cancelling tasks")
+                except Exception:
+                    # Recorded in _callback_errors; re-raised after cleanup below.
+                    pass

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	async def close(self, *, wait_callbacks: bool = True, timeout: Optional[float] = 10.0) -> None:
	callback_tasks = self.callback_tasks()
	if callback_tasks:
	if wait_callbacks and (timeout is None or timeout > 0):
	try:
	await self.wait_callbacks(timeout=timeout)
	except asyncio.TimeoutError:
	pass
	for task in callback_tasks:
	if not task.done():
	task.cancel()
	results = await asyncio.gather(*callback_tasks, return_exceptions=True)
	self._collect_callback_errors(results)
	for segment in self._segments:
	await segment.close()
	if self._sub is not None:
	await self._sub.close()
	if self._callback_errors:
	raise self._callback_errors[0]
	async def close(self, *, wait_callbacks: bool = True, timeout: Optional[float] = 10.0) -> None:
	callback_tasks = self.callback_tasks()
	if callback_tasks:
	if wait_callbacks and (timeout is None or timeout > 0):
	try:
	await self.wait_callbacks(timeout=timeout)
	except asyncio.TimeoutError:
	_LOG.debug("MediaOutput callback wait timed out; cancelling tasks")
	except Exception:
	# Recorded in _callback_errors; re-raised after cleanup below.
	pass
	for task in callback_tasks:
	if not task.done():
	task.cancel()
	results = await asyncio.gather(*callback_tasks, return_exceptions=True)
	self._collect_callback_errors(results)
	for segment in self._segments:
	await segment.close()
	if self._sub is not None:
	await self._sub.close()
	if self._callback_errors:
	raise self._callback_errors[0]

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/livepeer_gateway/media_output.py` around lines 549 - 567, Update
MediaOutput.close around wait_callbacks so callback-consumer errors are
collected rather than allowed to abort cleanup; handle the failure from
wait_callbacks alongside timeout handling, then continue cancelling remaining
callback tasks, closing all segments, and closing self._sub before propagating
the collected callback error.

[coderabbitai]

🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win

Livepeer-Segment can be None, which breaks the aiohttp request.

_payment_request explicitly allows segCreds to be absent (line 302-306 only type-checks when non-None), so payment.seg_creds may be None here. aiohttp rejects non-str header values with a TypeError, which is not caught by the ClientError/TimeoutError handlers below and escapes as an untyped error. The sync PaymentSession.send_payment already guards this with p.seg_creds or "" (line 475).

🐛 Proposed fix

         headers = {
             "Livepeer-Payment": payment.payment,
-            "Livepeer-Segment": payment.seg_creds,
+            "Livepeer-Segment": payment.seg_creds or "",
         }

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	headers = {
	"Livepeer-Payment": payment.payment,
	"Livepeer-Segment": payment.seg_creds,
	}
	headers = {
	"Livepeer-Payment": payment.payment,
	"Livepeer-Segment": payment.seg_creds or "",
	}

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/livepeer_gateway/remote_signer.py` around lines 255 - 258, Update the
headers construction in _payment_request so Livepeer-Segment always receives a
string when payment.seg_creds is None, matching the existing
PaymentSession.send_payment behavior by using an empty-string fallback.

[coderabbitai]

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

Empty orchestrators blocks fallback to discovery_url/signer_url.

The check is orchestrators is not None, so [] or "" takes the orchestrator branch, yields zero discovery URLs, and runner_selector raises NoRunnerAvailableError instead of falling through. discover_orchestrators in src/livepeer_gateway/discovery.py deliberately falls through on empty/whitespace-only input, and start_scope's documented precedence (scope.py lines 38-43) assumes the same. Normalize first and only take the orchestrator path when it actually yields URLs.

🐛 Proposed fix

-    if orchestrators is not None:
-        entries = await discover_orchestrator_runners(
-            orchestrators,
-            app=app,
-            gpu=gpu,
-        )
-    else:
+    entries: list[dict[str, Any]] = []
+    if orchestrators is not None and orchestrator_discovery_urls(orchestrators):
+        entries = await discover_orchestrator_runners(
+            orchestrators,
+            app=app,
+            gpu=gpu,
+        )
+    else:
         entries = await discover_runners(

(orchestrator_discovery_urls is already exported from .discovery.)

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	if orchestrators is not None:
	entries = await discover_orchestrator_runners(
	orchestrators,
	app=app,
	gpu=gpu,
	)
	else:
	entries = await discover_runners(
	signer_url=signer_url,
	signer_headers=signer_headers,
	discovery_url=discovery_url,
	discovery_headers=discovery_headers,
	app=app,
	gpu=gpu,
	)
	entries: list[dict[str, Any]] = []
	if orchestrators is not None and orchestrator_discovery_urls(orchestrators):
	entries = await discover_orchestrator_runners(
	orchestrators,
	app=app,
	gpu=gpu,
	)
	else:
	entries = await discover_runners(
	signer_url=signer_url,
	signer_headers=signer_headers,
	discovery_url=discovery_url,
	discovery_headers=discovery_headers,
	app=app,
	gpu=gpu,
	)

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/livepeer_gateway/selection.py` around lines 242 - 256, Update the
selection logic around discover_orchestrator_runners and discover_runners to
normalize orchestrators through the existing orchestrator_discovery_urls helper,
treating None, empty collections, and whitespace-only values as absent. Only use
the orchestrator path when normalization yields URLs; otherwise fall back to
discover_runners with the configured discovery_url and signer_url.

[coderabbitai]

Actionable comments posted: 2

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@src/livepeer_gateway/http.py`:
- Around line 326-327: Update open_stream to use aiohttp’s default TLS
certificate and hostname verification instead of constructing its session with
TCPConnector(ssl=False). If insecure TLS is required, thread an explicit opt-in
verification/SSL parameter consistently through open_stream, request_json, and
request_json_sync, preserving verified TLS by default.
- Around line 328-339: Update the request flow around session.request and the
resp.status >= 400 branch to guarantee both response release and session closure
via try/finally, including when reading resp.text() fails. Broaden request
exception handling to clean up and preserve the documented catch-all behavior
for unexpected exceptions, while retaining the existing LivepeerGatewayError
conversion and HTTP JSON error handling.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: 8a2a989b-4afd-444f-a33c-5cbeacf9e6cb

📥 Commits

Reviewing files that changed from the base of the PR and between 89ac0f7 and 9f2bc20.

📒 Files selected for processing (3)

• src/livepeer_gateway/__init__.py
• src/livepeer_gateway/http.py
• src/livepeer_gateway/live_runner.py

🚧 Files skipped from review as they are similar to previous changes (1)

• src/livepeer_gateway/live_runner.py

[coderabbitai]

🔒 Security & Privacy | 🟠 Major | ⚡ Quick win

open_stream disables TLS certificate/hostname verification (same pattern already flagged in this file).

aiohttp.TCPConnector(ssl=False) skips certificate and hostname validation for the entire streaming session, extending the previously-flagged TLS-bypass pattern (request_json_sync/request_json) to the new streaming path. This is a third code path in the file that unconditionally disables TLS verification instead of making it explicit/configurable.

🔒 Proposed fix: default to verified TLS, thread an explicit opt-out

-    timeout = aiohttp.ClientTimeout(total=None, sock_connect=connect_timeout, sock_read=None)
-    session = aiohttp.ClientSession(timeout=timeout, connector=aiohttp.TCPConnector(ssl=False))
+    timeout = aiohttp.ClientTimeout(total=None, sock_connect=connect_timeout, sock_read=None)
+    session = aiohttp.ClientSession(timeout=timeout)

If self-signed orchestrator/runner certificates genuinely need to be tolerated, surface that as an explicit, opt-in parameter shared with request_json/request_json_sync rather than disabling verification unconditionally.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/livepeer_gateway/http.py` around lines 326 - 327, Update open_stream to
use aiohttp’s default TLS certificate and hostname verification instead of
constructing its session with TCPConnector(ssl=False). If insecure TLS is
required, thread an explicit opt-in verification/SSL parameter consistently
through open_stream, request_json, and request_json_sync, preserving verified
TLS by default.

[coderabbitai]

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

Session/response can leak if the error-body read fails.

In the resp.status >= 400 branch, raw = await resp.text() isn't guarded — if reading the error body itself raises (e.g. connection drop mid-read), resp.release()/session.close() never run and the session leaks. Separately, session.request(...) at line 329 is only guarded for aiohttp.ClientError/asyncio.TimeoutError; any other unexpected exception during the request also bypasses session.close(), unlike request_json's documented catch-all handling of unexpected failures.

🧹 Proposed fix: guarantee cleanup with try/finally

     if resp.status >= 400:
-        raw = await resp.text()
-        resp.release()
-        await session.close()
+        try:
+            raw = await resp.text()
+        finally:
+            resp.release()
+            await session.close()
         _raise_http_json_error(resp.status, url, raw, dict(resp.headers.items()))

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/livepeer_gateway/http.py` around lines 328 - 339, Update the request flow
around session.request and the resp.status >= 400 branch to guarantee both
response release and session closure via try/finally, including when reading
resp.text() fails. Broaden request exception handling to clean up and preserve
the documented catch-all behavior for unexpected exceptions, while retaining the
existing LivepeerGatewayError conversion and HTTP JSON error handling.