Skip to content

Bump Livepeer version and enhance streaming functionality #2

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
eliteprox wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Bump Livepeer version and enhance streaming functionality #2

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
2a98429
bump livepeer-gateway
eliteprox Feb 5, 2026
451c8e2
chore: bump version to 1.0.4 and update livepeer-gateway dependency
eliteprox Mar 31, 2026
a4c412f
feat: enhance Livepeer streaming functionality with new nodes and aut…
eliteprox Apr 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion nodes/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
import logging
import sys

from . import frame_nodes, stream_url_node, pipeline_config_sdxl, js
from . import frame_nodes, stream_url_node, pipeline_config_sdxl, js, trickle_start_preview_node
from .stream import auth_routes, viewer_routes # noqa: F401


def _configure_logging():
Expand Down Expand Up @@ -30,6 +31,8 @@ def _configure_logging():
NODE_DISPLAY_NAME_MAPPINGS.update(pipeline_config_sdxl.NODE_DISPLAY_NAME_MAPPINGS)
NODE_CLASS_MAPPINGS.update(js.NODE_CLASS_MAPPINGS)
NODE_DISPLAY_NAME_MAPPINGS.update(js.NODE_DISPLAY_NAME_MAPPINGS)
NODE_CLASS_MAPPINGS.update(trickle_start_preview_node.NODE_CLASS_MAPPINGS)
NODE_DISPLAY_NAME_MAPPINGS.update(trickle_start_preview_node.NODE_DISPLAY_NAME_MAPPINGS)


__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS"]
Expand Down
134 changes: 102 additions & 32 deletions nodes/frame_nodes.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,13 +5,14 @@
import uuid
from dataclasses import dataclass
from typing import Any, Dict, Optional
from urllib.parse import urlparse

import folder_paths
import numpy as np
import torch
import cv2

from .stream.credentials import resolve_network_config
from .stream.credentials import BILLING_URL, cached_auth_status, resolve_orchestrator
from .stream.frame_bridge import enqueue_tensor_frame, enqueue_tensor_batch, queue_depth, has_loop, FRAME_BRIDGE
from .stream.network_controller import NetworkController, NetworkControllerConfig
from .stream.network_subscriber import NetworkSubscriber, NetworkSubscriberConfig
Expand All @@ -35,6 +36,47 @@ class _NetworkRuntime:
_NETWORK_RUNTIME = _NetworkRuntime()


def _format_device_auth_dialog(
device_auth: Dict[str, Any],
startup_error: str,
) -> str:
auth_url = str(device_auth.get("auth_url", "")).strip()
user_code = str(device_auth.get("user_code", "")).strip()
try:
expires_in = int(device_auth.get("expires_in", 0) or 0)
except (TypeError, ValueError):
expires_in = 0
expires_minutes = max(1, round(expires_in / 60)) if expires_in else None

verify_url = auth_url
if auth_url:
parsed = urlparse(auth_url)
if parsed.scheme and parsed.netloc:
verify_url = f"{parsed.scheme}://{parsed.netloc}/oidc/device"

details = [
"Livepeer login required (device code flow).",
f"1) Open: {auth_url or verify_url}",
f"2) Enter code: {user_code or '(missing code)'}",
"3) Approve access in your browser.",
"4) Re-run this workflow after completing login.",
]
if verify_url and verify_url != auth_url:
details.append(f"Fallback URL: {verify_url}")
if expires_minutes:
details.append(f"Code expires in about {expires_minutes} minute(s).")

error_lower = startup_error.lower()
if "expired_token" in error_lower or "expired" in error_lower:
details.append("The code expired. Run StartTrickleStream again to get a new code.")
elif "access_denied" in error_lower:
details.append("Authorization was denied. Retry and approve access in the browser.")
elif "authorization_pending" in error_lower:
details.append("Authorization is still pending. Complete login and run again.")

return "\n".join(details)


# --- Capture nodes ---


Expand Down Expand Up @@ -66,7 +108,7 @@ def INPUT_TYPES(cls) -> Dict[str, Any]:
RETURN_TYPES = ("IMAGE",)
RETURN_NAMES = ("images",)
FUNCTION = "batch_frames"
CATEGORY = "Trickle"
CATEGORY = "Livepeer/Trickle"
OUTPUT_NODE = False

@classmethod
Expand Down Expand Up @@ -188,20 +230,16 @@ class TrickleConfig:
"""
Configuration node for trickle streaming parameters.
Outputs a config dict that can be connected to Start Trickle Stream.

Authentication (billing_url / client_id) is handled automatically
via package-level constants. Orchestrator discovery is used by
default; set orchestrator_url only to pin a specific orchestrator.
"""

@classmethod
def INPUT_TYPES(cls) -> Dict[str, Any]:
return {
"required": {
"orchestrator_url": ("STRING", {
"default": "https://localhost:8935",
"tooltip": "Orchestrator URL (e.g., https://hky.eliteencoder.net:8936)",
}),
"signer_url": ("STRING", {
"default": "https://signer.eliteencoder.net",
"tooltip": "Signer URL for authentication",
}),
"model_id": ("STRING", {
"default": "noop",
"tooltip": "Model ID to use (e.g., noop, comfystream)",
Expand All @@ -222,6 +260,10 @@ def INPUT_TYPES(cls) -> Dict[str, Any]:
}),
},
"optional": {
"orchestrator_url": ("STRING", {
"default": "",
"tooltip": "Pin a specific orchestrator (e.g., https://hky.eliteencoder.net:8936). Leave empty for automatic discovery.",
}),
"pipeline_params": ("DICT", {
"tooltip": "Connect pipeline config (e.g., StreamDiffusion SDXL params_dict output)",
}),
Expand All @@ -231,26 +273,24 @@ def INPUT_TYPES(cls) -> Dict[str, Any]:
RETURN_TYPES = ("TRICKLE_CONFIG",)
RETURN_NAMES = ("config",)
FUNCTION = "create_config"
CATEGORY = "Trickle"
CATEGORY = "Livepeer/Trickle"

def create_config(
self,
orchestrator_url: str,
signer_url: str,
model_id: str,
fps: float,
keyframe_interval: float,
orchestrator_url: str = "",
pipeline_params: Dict[str, Any] = None,
) -> tuple:
config = {
"orchestrator_url": orchestrator_url,
"signer_url": signer_url,
"model_id": model_id,
"fps": fps,
"keyframe_interval": keyframe_interval,
"orchestrator_url": orchestrator_url,
"pipeline_params": pipeline_params or {},
}

return (config,)


Expand Down Expand Up @@ -295,7 +335,7 @@ def INPUT_TYPES(cls) -> Dict[str, Any]:
RETURN_TYPES = ("STRING", "STRING", "STRING", "STRING")
RETURN_NAMES = ("manifest_id", "publish_url", "subscribe_url", "error")
FUNCTION = "start_trickle_stream"
CATEGORY = "Trickle"
CATEGORY = "Livepeer/Trickle"
OUTPUT_NODE = True

@classmethod
Expand Down Expand Up @@ -427,26 +467,19 @@ def start_trickle_stream(
# Proactively check for stale/dead streams before proceeding
self._check_and_clear_stale_stream()

# Extract values from config
orchestrator_url = config.get("orchestrator_url", "https://localhost:8935")
signer_url = config.get("signer_url", "")
model_id = config.get("model_id", "noop")
fps = config.get("fps", 30.0)
keyframe_interval = config.get("keyframe_interval", 2.0)

# Pipeline params passed through from TrickleConfig
orchestrator_url = config.get("orchestrator_url", "")
pipeline_params = config.get("pipeline_params", {})

resolved_orch, resolved_signer = resolve_network_config(orchestrator_url, signer_url)
controller_config = NetworkControllerConfig(
orchestrator_url=resolved_orch,
signer_url=resolved_signer or None,
model_id=model_id,
fps=float(fps),
frame_width=width,
frame_height=height,
keyframe_interval_s=float(keyframe_interval),

orchestrator_url=resolve_orchestrator(orchestrator_url),
)
controller = _get_controller(controller_config)
Comment on lines 476 to 484

@coderabbitai coderabbitai Bot Apr 20, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Run the auth check before creating/stashing a controller.

_get_controller() is called before cached_auth_status(), so an unauthenticated start leaves an IDLE controller in _NETWORK_RUNTIME.controller. Downstream nodes then stop hitting the last_startup_error path and just skip frames as if a controller exists. Move the auth gate above _get_controller() or clear _NETWORK_RUNTIME.controller before returning the auth error.

Also applies to: 539-553

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@nodes/frame_nodes.py` around lines 476 - 484, The code currently calls
_get_controller(controller_config) before checking cached_auth_status(), which
can leave an IDLE controller stored in _NETWORK_RUNTIME.controller and cause
downstream nodes to skip the last_startup_error path; move the
cached_auth_status() auth gate so it runs before calling _get_controller(), or
if you prefer minimal changes clear _NETWORK_RUNTIME.controller (set to None)
immediately before returning the auth error; update both locations (the block
using controller_config/_get_controller at lines around the shown snippet and
the similar block at 539-553) so the auth check precedes controller creation or
ensure the controller is cleared on auth failure.


Expand Down Expand Up @@ -503,6 +536,21 @@ def start_trickle_stream(
# during startup. The subscriber handles resetting when the URL changes (i.e.,
# when connecting to a genuinely different stream). For same-stream reconnects,
# keeping the last frame is better UX than showing black.
has_cached_auth, auth_reason = cached_auth_status()
if not has_cached_auth:
error_msg = (
"Livepeer login required before starting stream.\n"
f"{auth_reason}\n"
"Click Livepeer Login in Settings (Livepeer -> Authentication -> Login), "
"then re-run StartTrickleStream.\n"
f"Billing URL: {BILLING_URL}"
)
_NETWORK_RUNTIME.last_startup_error = error_msg
self._status_cache = ("", "", "", error_msg)
return {
"ui": {"text": [f"ERROR: {error_msg}"]},
"result": self._status_cache,
}

try:
status = controller.start(
Expand Down Expand Up @@ -541,6 +589,15 @@ def start_trickle_stream(
error_str = str(exc)
LOGGER.error("StartTrickleStream: Failed to start stream: %s", error_str)

# Check if OIDC device auth was pending when the error occurred
device_auth = NetworkController._device_auth_info
if device_auth:
error_str = _format_device_auth_dialog(
device_auth=device_auth,
startup_error=error_str,
)
NetworkController._device_auth_info = None

# Properly stop and clean up the controller before clearing
if controller:
try:
Expand Down Expand Up @@ -622,7 +679,7 @@ def INPUT_TYPES(cls) -> Dict[str, Any]:
RETURN_TYPES = ("STRING", "STRING", "STRING", "STRING")
RETURN_NAMES = ("manifest_id", "publish_url", "subscribe_url", "error")
FUNCTION = "start_stream_from_video"
CATEGORY = "Trickle"
CATEGORY = "Livepeer/Trickle"
OUTPUT_NODE = True

@classmethod
Expand Down Expand Up @@ -773,7 +830,7 @@ def INPUT_TYPES(cls) -> Dict[str, Any]:
RETURN_TYPES = ()
RETURN_NAMES = ()
FUNCTION = "push_frame"
CATEGORY = "Trickle"
CATEGORY = "Livepeer/Trickle"
OUTPUT_NODE = True

@classmethod
Expand Down Expand Up @@ -886,7 +943,7 @@ def INPUT_TYPES(cls) -> Dict[str, Any]:
RETURN_TYPES = ()
RETURN_NAMES = ()
FUNCTION = "feed_video"
CATEGORY = "Trickle"
CATEGORY = "Livepeer/Trickle"
OUTPUT_NODE = True

@classmethod
Expand Down Expand Up @@ -1004,6 +1061,8 @@ class TrickleFrameOutput:

# Class-level counter per node unique_id to ensure preview filenames change each execution
_execution_counters: Dict[str, int] = {}
INITIAL_FRAME_WAIT_SECONDS = 2.5
INITIAL_FRAME_POLL_SECONDS = 0.05

def __init__(self):
self._blank = self._blank_tensor()
Expand All @@ -1023,7 +1082,7 @@ def INPUT_TYPES(cls) -> Dict[str, Any]:
RETURN_TYPES = ("IMAGE",)
RETURN_NAMES = ("image",)
FUNCTION = "pull_frame"
CATEGORY = "Trickle"
CATEGORY = "Livepeer/Trickle"
OUTPUT_NODE = True # Always executes and shows preview

@classmethod
Expand All @@ -1050,6 +1109,17 @@ def pull_frame(self, unique_id: str = ""):

try:
frame_np, timestamp, has_frame = TRICKLE_OUTPUT_BRIDGE.get_frame_or_blank_sync()
if not has_frame and subscriber.running and subscriber.task_alive:
# First-frame warmup: wait briefly so users do not need to run twice
# after starting or restarting a stream.
deadline = time.perf_counter() + self.INITIAL_FRAME_WAIT_SECONDS
while time.perf_counter() < deadline:
frame_np, timestamp, has_frame = TRICKLE_OUTPUT_BRIDGE.get_frame_or_blank_sync()
if has_frame:
break
if not subscriber.running or not subscriber.task_alive:
break
time.sleep(self.INITIAL_FRAME_POLL_SECONDS)
tensor = torch.from_numpy(frame_np.astype(np.float32) / 255.0).unsqueeze(0)
return self._return_with_preview(tensor, unique_id)
except Exception as exc:
Expand Down Expand Up @@ -1120,7 +1190,7 @@ def INPUT_TYPES(cls) -> Dict[str, Any]:
RETURN_TYPES = ()
RETURN_NAMES = ()
FUNCTION = "update_trickle_stream"
CATEGORY = "Trickle"
CATEGORY = "Livepeer/Trickle"
OUTPUT_NODE = True

@classmethod
Expand All @@ -1136,7 +1206,7 @@ def update_trickle_stream(self, control_payload: Dict[str, Any], enabled: bool =
return ()
try:
future = asyncio.run_coroutine_threadsafe(
controller.job.control.write_control(control_payload),
controller.job.control.write(control_payload),
controller.loop,
)
future.result(timeout=5)
Expand Down
Loading