update .env.example to include BRANDO_API_KEY and MINIO_ENDPOINT

feat: migrate ComfyUI nodes from presigned URLs to Brain API storage endpoints

## Core Architecture Changes
- Add new BrainApiClient for direct Brain API communication
- Replace presigned URL workflow with storage_id/filename pattern
- Update MediaStreamInput to use storage_id + filename instead of presigned_download_url
- Update MediaStreamOutput to use storage_id + filename instead of presigned_upload_url

## Brain API Client Implementation
- Add comprehensive BrainApiClient with authentication and error handling
- Implement get_presigned_upload_url() and get_presigned_download_url() methods
- Add proper Bearer token authentication for Brain API requests
- Include environment variable configuration for Brain API endpoint and credentials

## Media Stream Node Updates
- Refactor MediaStreamInput to request presigned URLs from Brain API using storage_id
- Update MediaStreamOutput to upload directly to MinIO using Brain API presigned URLs
- Simplify node interface by removing presigned URL inputs from workflow templates
- Update JavaScript frontend to hide system inputs and manage node UI state

## Infrastructure Improvements
- Add .idea/ to .gitignore for IDE file exclusions
- Update user_input.py to use new storage_id pattern
- Streamline web interface JavaScript for cleaner node management

.env.example

@@ -0,0 +1,30 @@
+# --- BRAIN CREDENTIALS ---
+BRANDO_API_KEY=...
+
+# --- S3 (AWS/MINIO) CREDENTIALS ---
+## These are used by both scripts to connect to SQS and S3 (MinIO).
+AWS_ACCESS_KEY_ID=minioadmin
+AWS_SECRET_ACCESS_KEY=...
+AWS_DEFAULT_REGION=us-east-1
+MINIO_ENDPOINT=http://127.0.0.1:9000
+
+# --- SQS SETTINGS ---
+## Toggles functionality for the SQS Worker Consumer
+SQS_ENABLED=false
+
+## For media_stream.py (MediaStreamOutput Node)
+### Endpoint for the SQS service where completion messages are sent.
+SQS_ENDPOINT_URL=http://127.0.0.1:9324
+
+### The specific SQS queue the worker should push job status updates to.
+SQS_JOB_STATUS_UPDATES_QUEUE_NAME=job_status_updates
+
+## For worker_consumer.py (Job Consumer)
+### The specific SQS queue which the worker should poll for new jobs.
+SQS_JOBS_TO_PROCESS_QUEUE_NAME=jobs_to_process
+
+## (Optional) For worker_consumer.py (Job Consumer)
+### The local URL of the ComfyUI API server.
+# You only need to set this if your ComfyUI server is NOT running on the default port 8188.
+# COMFYUI_API_URL=http://127.0.0.1:8188
+# COMFYUI_WS_URL=ws://127.0.0.1:8188

.gitignore

@@ -157,4 +157,4 @@ cython_debug/
 #  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
-#.idea/
+.idea/

README.md

@@ -2,6 +2,10 @@
 
 A collection of utility nodes for ComfyUI focusing on list manipulation, batch operations, and advanced I/O functionality.
 
+## Prerequisites
+
+- `comfyui-kjnodes` custom_nodes repo
+
 ## 🏭 Generators
 
 <details>
@@ -188,4 +192,31 @@ Uploads video files to a HuggingFace dataset.
 | filename_prefix | STRING | Prefix for saved files |
 
 **Notes**: Handles batch upload of multiple video files.
+</details>
+
+## 📡 Core Nilor Services
+
+<details>
+<summary><b>Worker Consumer Service</b></summary>
+
+The `worker_consumer.py` script is a background service that runs on each ComfyUI worker. It is responsible for pulling jobs from the central ElasticMQ `jobs_to_process` queue and submitting them to its local ComfyUI instance for processing. This service is essential for the distributed architecture of the system.
+
+**Key Responsibilities:**
+-   Continuously polls the `jobs_to_process` queue for new jobs using long polling.
+-   When a job is received, it extracts the workflow data and submits it to the local ComfyUI server.
+-   Deletes the job message from the queue upon successful submission to prevent reprocessing.
+-   If submission fails, the message remains on the queue to be picked up by another worker.
+
+</details>
+
+<details>
+<summary><b>Environment Variables</b></summary>
+
+The `nilor-nodes` require a `.env` file to be present in the `ComfyUI` directory to configure the connection to the core services (MinIO, ElasticMQ, and the Brain API). To set it up, create a file named `.env` in the root of your `ComfyUI` directory by copying the `.env.example` template.
+
+**Instructions:**
+1.  Create a new file named `.env` in the `ComfyUI` directory.
+2.  Copy the contents of the `.env.example` file into your new `.env` file.
+3.  Replace the placeholder values with your actual credentials and endpoint URLs for your local or production environment.
+
 </details>

__init__.py

@@ -1,3 +1,80 @@
-from .nilornodes import NODE_CLASS_MAPPINGS, NODE_DISPLAY_NAME_MAPPINGS
+import os
+import threading
+import asyncio
+from dotenv import load_dotenv
+
+# --- Nilor-Nodes Custom Node Registration and Startup ---
+# This file is executed when ComfyUI starts and discovers this custom node directory.
+# It's responsible for:
+# 1. Starting background services (like the SQS worker and a FastAPI server).
+# 2. Registering the custom nodes with ComfyUI so they appear in the menu.
+
+
+# --- Load Environment Variables ---
+# Get the directory of the current script
+current_dir = os.path.dirname(os.path.abspath(__file__))
+# Construct the path to the .env file
+dotenv_path = os.path.join(current_dir, ".env")
+# Load the .env file, overriding any pre-existing process env for these keys
+load_dotenv(dotenv_path=dotenv_path, override=True)
+
+
+# --- Background Services ---
+
+
+def start_consumer_loop():
+    """Synchronous wrapper to run the asyncio event loop for the consumer."""
+    from .worker_consumer import consume_jobs
+
+    asyncio.run(consume_jobs())
+
+
+# Start the SQS Worker Consumer (controlled by SQS_ENABLED)
+raw_sqs_enabled = os.getenv("SQS_ENABLED", "false")
+env_sqs_enabled = raw_sqs_enabled.strip().lower() == "true"
+if env_sqs_enabled:
+    consumer_thread = threading.Thread(target=start_consumer_loop, daemon=True)
+    consumer_thread.start()
+    print(
+        f"✅ Nilor-Nodes: SQS worker consumer thread started (SQS_ENABLED={raw_sqs_enabled} in .env)."
+    )
+else:
+    print(
+        f"⚠️ Nilor-Nodes: SQS worker consumer functionality is disabled (SQS_ENABLED={raw_sqs_enabled} in .env)."
+    )
+
+
+# --- Node Registration ---
+from .nilornodes import (
+    NODE_CLASS_MAPPINGS as base_NODE_CLASS_MAPPINGS,
+    NODE_DISPLAY_NAME_MAPPINGS as base_NODE_DISPLAY_NAME_MAPPINGS,
+)
+from .media_stream import (
+    NODE_CLASS_MAPPINGS as ms_NODE_CLASS_MAPPINGS,
+    NODE_DISPLAY_NAME_MAPPINGS as ms_NODE_DISPLAY_NAME_MAPPINGS,
+)
+from .user_input import (
+    NODE_CLASS_MAPPINGS as ui_NODE_CLASS_MAPPINGS,
+    NODE_DISPLAY_NAME_MAPPINGS as ui_NODE_DISPLAY_NAME_MAPPINGS,
+)
+from .controllers import (
+    NODE_CLASS_MAPPINGS as ctrl_NODE_CLASS_MAPPINGS,
+    NODE_DISPLAY_NAME_MAPPINGS as ctrl_NODE_DISPLAY_NAME_MAPPINGS,
+)
+
+NODE_CLASS_MAPPINGS = dict(base_NODE_CLASS_MAPPINGS)
+NODE_DISPLAY_NAME_MAPPINGS = dict(base_NODE_DISPLAY_NAME_MAPPINGS)
+
+NODE_CLASS_MAPPINGS.update(ms_NODE_CLASS_MAPPINGS)
+NODE_DISPLAY_NAME_MAPPINGS.update(ms_NODE_DISPLAY_NAME_MAPPINGS)
+
+NODE_CLASS_MAPPINGS.update(ui_NODE_CLASS_MAPPINGS)
+NODE_DISPLAY_NAME_MAPPINGS.update(ui_NODE_DISPLAY_NAME_MAPPINGS)
+
+NODE_CLASS_MAPPINGS.update(ctrl_NODE_CLASS_MAPPINGS)
+NODE_DISPLAY_NAME_MAPPINGS.update(ctrl_NODE_DISPLAY_NAME_MAPPINGS)
+
+WEB_DIRECTORY = "./web"
 
 __all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS"]
+print("✅ Nilor-Nodes: All custom nodes registered.")

brain_api_client.py

@@ -0,0 +1,392 @@
+"""
+Brain API Client for ComfyUI Nodes
+
+This client provides methods to interact with the Brain API storage endpoints,
+replacing the need for pre-signed URLs in the ComfyUI workflow.
+"""
+
+import requests
+import os
+import logging
+from typing import Optional, Dict, Any
+from dotenv import load_dotenv
+
+# Load environment variables
+current_dir = os.path.dirname(os.path.abspath(__file__))
+dotenv_path = os.path.join(current_dir, ".env")
+load_dotenv(dotenv_path=dotenv_path)
+
+# Setup logging
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+)
+
+
+class BrainApiClient:
+    """
+    Client for interacting with Brain API storage endpoints.
+
+    This client handles authentication and provides methods for uploading,
+    downloading, and deleting files through the Brain API storage endpoints.
+    """
+
+    def __init__(self):
+        """Initialize the Brain API client with configuration from environment variables."""
+        self.base_url = os.getenv(
+            "BRANDO_BRAIN_API_BASE_URL", "http://localhost:2024/api"
+        )
+        self.api_key = os.getenv("BRANDO_API_KEY")
+
+        if not self.api_key:
+            raise ValueError(
+                "BRANDO_API_KEY environment variable is required for Brain API authentication"
+            )
+
+        self.headers = {
+            "Authorization": f"Bearer {self.api_key}",
+            "User-Agent": "ComfyUI-NilorNodes/1.0",
+        }
+
+        logging.info(f"Brain API Client initialized with base URL: {self.base_url}")
+
+    def upload_file_to_storage(self, file_path: str, filename: str) -> Dict[str, Any]:
+        """
+        Upload a file to Brain API storage and return storage metadata.
+
+        Args:
+            file_path: Local path to the file to upload
+            filename: Name to use for the uploaded file
+
+        Returns:
+            Dict containing storage_id and filename
+
+        Raises:
+            requests.RequestException: If upload fails
+            FileNotFoundError: If file_path doesn't exist
+        """
+        if not os.path.exists(file_path):
+            raise FileNotFoundError(f"File not found: {file_path}")
+
+        url = f"{self.base_url}/storage/upload"
+
+        try:
+            with open(file_path, "rb") as file:
+                files = {"file": (filename, file, "application/octet-stream")}
+
+                logging.info(f"Uploading file '{filename}' to Brain API storage...")
+                response = requests.post(
+                    url, files=files, headers=self.headers, timeout=300
+                )
+                response.raise_for_status()
+
+                result = response.json()
+                logging.info(
+                    f"Upload successful. Storage ID: {result.get('storage_id')}"
+                )
+                return result
+
+        except requests.RequestException as e:
+            logging.error(f"Failed to upload file '{filename}': {e}")
+            raise
+        except Exception as e:
+            logging.error(f"Unexpected error uploading file '{filename}': {e}")
+            raise
+
+    def upload_fileobj_to_storage(
+        self, file_obj, filename: str, content_type: str = "application/octet-stream"
+    ) -> Dict[str, Any]:
+        """
+        Upload a file-like object to Brain API storage and return storage metadata.
+
+        Args:
+            file_obj: File-like object to upload
+            filename: Name to use for the uploaded file
+            content_type: MIME type of the file
+
+        Returns:
+            Dict containing storage_id and filename
+
+        Raises:
+            requests.RequestException: If upload fails
+        """
+        url = f"{self.base_url}/storage/upload"
+
+        try:
+            files = {"file": (filename, file_obj, content_type)}
+
+            logging.info(f"Uploading file object '{filename}' to Brain API storage...")
+            response = requests.post(
+                url, files=files, headers=self.headers, timeout=300
+            )
+            response.raise_for_status()
+
+            result = response.json()
+            logging.info(f"Upload successful. Storage ID: {result.get('storage_id')}")
+            return result
+
+        except requests.RequestException as e:
+            logging.error(f"Failed to upload file object '{filename}': {e}")
+            raise
+        except Exception as e:
+            logging.error(f"Unexpected error uploading file object '{filename}': {e}")
+            raise
+
+    def download_file_from_storage(
+        self, storage_id: str, filename: str, dest_path: str
+    ) -> str:
+        """
+        Download a file from Brain API storage to a local path.
+
+        Args:
+            storage_id: Storage ID of the file to download
+            filename: Name of the file to download
+            dest_path: Local path where the file should be saved
+
+        Returns:
+            Path to the downloaded file
+
+        Raises:
+            requests.RequestException: If download fails
+        """
+        url = f"{self.base_url}/storage/{storage_id}"
+        params = {"filename": filename}
+
+        try:
+            logging.info(
+                f"Downloading file '{filename}' (storage_id: {storage_id}) from Brain API storage..."
+            )
+            response = requests.get(
+                url, params=params, headers=self.headers, timeout=300, stream=True
+            )
+            response.raise_for_status()
+
+            # Ensure destination directory exists
+            os.makedirs(os.path.dirname(dest_path), exist_ok=True)
+
+            with open(dest_path, "wb") as f:
+                for chunk in response.iter_content(chunk_size=8192):
+                    f.write(chunk)
+
+            logging.info(f"Download successful. File saved to: {dest_path}")
+            return dest_path
+
+        except requests.RequestException as e:
+            logging.error(
+                f"Failed to download file '{filename}' (storage_id: {storage_id}): {e}"
+            )
+            raise
+        except Exception as e:
+            logging.error(f"Unexpected error downloading file '{filename}': {e}")
+            raise
+
+    def get_file_from_storage(self, storage_id: str, filename: str) -> bytes:
+        """
+        Get file content from Brain API storage as bytes.
+
+        Args:
+            storage_id: Storage ID of the file to download
+            filename: Name of the file to download
+
+        Returns:
+            File content as bytes
+
+        Raises:
+            requests.RequestException: If download fails
+        """
+        url = f"{self.base_url}/storage/{storage_id}"
+        params = {"filename": filename}
+
+        try:
+            logging.info(
+                f"Getting file '{filename}' (storage_id: {storage_id}) from Brain API storage..."
+            )
+            response = requests.get(
+                url, params=params, headers=self.headers, timeout=300
+            )
+            response.raise_for_status()
+
+            logging.info(
+                f"File retrieval successful. Size: {len(response.content)} bytes"
+            )
+            return response.content
+
+        except requests.RequestException as e:
+            logging.error(
+                f"Failed to get file '{filename}' (storage_id: {storage_id}): {e}"
+            )
+            raise
+        except Exception as e:
+            logging.error(f"Unexpected error getting file '{filename}': {e}")
+            raise
+
+    def delete_file_from_storage(self, storage_id: str, filename: str) -> None:
+        """
+        Delete a file from Brain API storage.
+
+        Args:
+            storage_id: Storage ID of the file to delete
+            filename: Name of the file to delete
+
+        Raises:
+            requests.RequestException: If deletion fails
+        """
+        url = f"{self.base_url}/storage/{storage_id}"
+        params = {"filename": filename}
+
+        try:
+            logging.info(
+                f"Deleting file '{filename}' (storage_id: {storage_id}) from Brain API storage..."
+            )
+            response = requests.delete(
+                url, params=params, headers=self.headers, timeout=60
+            )
+            response.raise_for_status()
+
+            logging.info(f"File deletion successful")
+
+        except requests.RequestException as e:
+            logging.error(
+                f"Failed to delete file '{filename}' (storage_id: {storage_id}): {e}"
+            )
+            raise
+        except Exception as e:
+            logging.error(f"Unexpected error deleting file '{filename}': {e}")
+            raise
+
+    def get_presigned_upload_url(
+        self, filename: str, content_type: str, minio_endpoint: str
+    ) -> Dict[str, Any]:
+        """
+        Get a presigned upload URL from Brain API for direct MinIO upload.
+
+        Args:
+            filename: Name of the file to upload
+            content_type: MIME type of the file
+            minio_endpoint: MinIO endpoint that ComfyUI can access
+
+        Returns:
+            Dict containing storage_id, upload_url, and object_key
+
+        Raises:
+            requests.RequestException: If request fails
+        """
+        url = f"{self.base_url}/storage/generate-upload-url"
+        payload = {
+            "filename": filename,
+            "content_type": content_type,
+            "minio_endpoint": minio_endpoint,
+        }
+
+        try:
+            logging.info(
+                f"Requesting presigned upload URL for '{filename}' from Brain API..."
+            )
+            response = requests.post(
+                url, json=payload, headers=self.headers, timeout=30
+            )
+            response.raise_for_status()
+
+            result = response.json()
+            logging.info(
+                f"Presigned upload URL generated. Storage ID: {result.get('storage_id')}"
+            )
+            return result
+
+        except requests.RequestException as e:
+            logging.error(f"Failed to get presigned upload URL for '{filename}': {e}")
+            raise
+        except Exception as e:
+            logging.error(
+                f"Unexpected error getting presigned upload URL for '{filename}': {e}"
+            )
+            raise
+
+    def get_presigned_download_url(
+        self, storage_id: str, filename: str, minio_endpoint: str
+    ) -> Dict[str, Any]:
+        """
+        Get a presigned download URL from Brain API for direct MinIO download.
+
+        Args:
+            storage_id: Storage ID of the file to download
+            filename: Name of the file to download
+            minio_endpoint: MinIO endpoint that ComfyUI can access
+
+        Returns:
+            Dict containing download_url
+
+        Raises:
+            requests.RequestException: If request fails
+        """
+        url = f"{self.base_url}/storage/generate-download-url"
+        payload = {
+            "storage_id": storage_id,
+            "filename": filename,
+            "minio_endpoint": minio_endpoint,
+        }
+
+        try:
+            logging.info(
+                f"Requesting presigned download URL for '{filename}' (storage_id: {storage_id}) from Brain API..."
+            )
+            response = requests.post(
+                url, json=payload, headers=self.headers, timeout=30
+            )
+            response.raise_for_status()
+
+            result = response.json()
+            logging.info(f"Presigned download URL generated for '{filename}'")
+            return result
+
+        except requests.RequestException as e:
+            logging.error(
+                f"Failed to get presigned download URL for '{filename}' (storage_id: {storage_id}): {e}"
+            )
+            raise
+        except Exception as e:
+            logging.error(
+                f"Unexpected error getting presigned download URL for '{filename}': {e}"
+            )
+            raise
+
+    def health_check(self) -> bool:
+        """
+        Check if the Brain API is accessible and authentication is working.
+
+        Returns:
+            True if API is accessible, False otherwise
+        """
+        try:
+            # Try to access a simple endpoint to verify connectivity
+            url = f"{self.base_url}/health"  # Assuming there's a health endpoint
+            response = requests.get(url, headers=self.headers, timeout=10)
+            return response.status_code == 200
+        except:
+            # If health endpoint doesn't exist, try the storage upload endpoint
+            # with a HEAD request to check authentication
+            try:
+                url = f"{self.base_url}/storage/upload"
+                response = requests.head(url, headers=self.headers, timeout=10)
+                return response.status_code in [
+                    200,
+                    405,
+                ]  # 405 Method Not Allowed is OK for HEAD
+            except:
+                return False
+
+
+# Global client instance
+_brain_api_client = None
+
+
+def get_brain_api_client() -> BrainApiClient:
+    """
+    Get or create the global Brain API client instance.
+
+    Returns:
+        BrainApiClient instance
+    """
+    global _brain_api_client
+    if _brain_api_client is None:
+        _brain_api_client = BrainApiClient()
+    return _brain_api_client

controllers.py

@@ -0,0 +1,88 @@
+category = "Nilor Nodes 👺"
+subcategories = {
+    "io": "/IO",
+}
+
+# Unique hook type for controller wiring (used by both Preset and Group controllers)
+CONTROLLER_HOOK = "CONTROLLER_HOOK"
+
+
+class NilorPreset:
+    """
+    Declarative controller that binds a Brando preset group to a set of connected inputs.
+
+    - preset_group_name: Semantic key used to look up choices and values in
+      presets_config.json5 via PresetsService (drives dropdown + value application).
+    - _preset_hook_*: Dynamic inputs that accept CONTROLLER_HOOK from NilorUserInput_* `_controller_hook` outputs.
+    """
+
+    @classmethod
+    def INPUT_TYPES(cls):
+        # Start with a single hook; dynamic inputs handled by companion JS
+        optional_inputs = {"_preset_hook_1": (CONTROLLER_HOOK,)}
+        return {
+            "required": {
+                # Lookup key in presets_config.json5 (NOT a UI label)
+                "preset_group_name": (
+                    "STRING",
+                    {"default": "my_preset", "multiline": False},
+                ),
+            },
+            "optional": optional_inputs,
+        }
+
+    # No outputs; declarative controller only
+    RETURN_TYPES = tuple()
+    RETURN_NAMES = tuple()
+    FUNCTION = "do_nothing"
+    CATEGORY = category + subcategories["io"]
+    OUTPUT_NODE = True
+
+    def do_nothing(self, **kwargs):
+        # This node performs no computation; it exists for declarative wiring only
+        return tuple()
+
+
+class NilorGroup:
+    """
+    Declarative UI-grouper that clusters connected inputs together in the Brando UI.
+
+    - group_label: Purely a visual label for the gr.Group that will contain the inputs.
+      It does NOT look up presets or apply values.
+    - _group_hook_*: Dynamic inputs that accept CONTROLLER_HOOK from NilorUserInput_* `_controller_hook` outputs.
+      Reuses a shared controller hook so no additional output types are required on input nodes.
+    """
+
+    @classmethod
+    def INPUT_TYPES(cls):
+        # Start with a single hook; dynamic inputs handled by companion JS
+        optional_inputs = {"_group_hook_1": (CONTROLLER_HOOK,)}
+        return {
+            "required": {
+                # UI label only (NOT used to look up presets)
+                "group_label": ("STRING", {"default": "my_group", "multiline": False}),
+            },
+            "optional": optional_inputs,
+        }
+
+    # No outputs; declarative controller only
+    RETURN_TYPES = tuple()
+    RETURN_NAMES = tuple()
+    FUNCTION = "do_nothing"
+    CATEGORY = category + subcategories["io"]
+    OUTPUT_NODE = True
+
+    def do_nothing(self, **kwargs):
+        # Declarative only
+        return tuple()
+
+
+NODE_CLASS_MAPPINGS = {
+    "NilorPreset": NilorPreset,
+    "NilorGroup": NilorGroup,
+}
+
+NODE_DISPLAY_NAME_MAPPINGS = {
+    "NilorPreset": "👺 User Input Preset Controller",
+    "NilorGroup": "👺 User Input Group Controller",
+}

media_stream.py

@@ -0,0 +1,449 @@
+import torch
+import numpy as np
+from PIL import Image
+import requests
+import io
+import logging
+import imageio.v2 as imageio
+import mimetypes
+import boto3
+import os
+import json
+from dotenv import load_dotenv
+from .brain_api_client import get_brain_api_client
+
+# --- Load Environment Variables ---
+# Get the directory of the current script
+current_dir = os.path.dirname(os.path.abspath(__file__))
+# Construct the path to the .env file
+dotenv_path = os.path.join(current_dir, ".env")
+# Load the .env file
+load_dotenv(dotenv_path=dotenv_path)
+
+# --- Setup Logging ---
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+)
+
+# --- Node Categories ---
+category = "Nilor Nodes 👺"
+subcategories = {
+    "streaming": "/Streaming",
+}
+
+
+# --- MediaStreamInput: Universal Media Downloader ---
+class MediaStreamInput:
+    """
+    A custom node to download an image/video from a pre-signed URL and provide it as a tensor.
+    """
+
+    def __init__(self):
+        pass
+
+    @classmethod
+    def INPUT_TYPES(s):
+        return {
+            "required": {
+                "input_name": (
+                    "STRING",
+                    {"default": "default_input", "multiline": False},
+                ),
+                "format": (["image", "image_batch", "video"],),
+                "storage_id": (
+                    "STRING",
+                    {"default": "<auto-filled by system>", "multiline": False},
+                ),
+                "filename": (
+                    "STRING",
+                    {"default": "<auto-filled by system>", "multiline": False},
+                ),
+            },
+            "hidden": {},
+        }
+
+    RETURN_TYPES = ("IMAGE",)
+    RETURN_NAMES = ("image",)
+    FUNCTION = "download"
+    CATEGORY = category + subcategories["streaming"]
+
+    def download(
+        self,
+        storage_id: str,
+        filename: str,
+        format: str,
+        input_name: str = "default_input",
+    ):
+        logging.info(
+            f"ℹ️\u2009 Nilor-Nodes: MediaStreamInput: Downloading file '{filename}' (storage_id: {storage_id}) for input '{input_name}' with format '{format}'"
+        )
+        try:
+            # Get Brain API client and MinIO endpoint
+            brain_client = get_brain_api_client()
+            minio_endpoint = os.getenv("MINIO_ENDPOINT")
+            if not minio_endpoint:
+                raise ValueError(
+                    "MINIO_ENDPOINT environment variable is required but not set"
+                )
+
+            # Two-phase download for batches: manifest first, then assets
+            if format == "image_batch":
+                # Get presigned download URL for manifest
+                manifest_url_response = brain_client.get_presigned_download_url(
+                    storage_id, filename, minio_endpoint
+                )
+                manifest_url = manifest_url_response["download_url"]
+
+                # Download manifest file directly from MinIO
+                manifest_response = requests.get(manifest_url, timeout=300)
+                manifest_response.raise_for_status()
+                manifest = json.loads(manifest_response.content.decode("utf-8"))
+
+                logging.info(
+                    f"ℹ️\u2009 Nilor-Nodes: Processing manifest for '{manifest.get('input_name')}' with {len(manifest.get('files', []))} assets."
+                )
+
+                # Sort files by sequence number to ensure correct order
+                sorted_files = sorted(
+                    manifest.get("files", []), key=lambda x: x.get("sequence", 0)
+                )
+
+                # Download all assets using presigned URLs
+                asset_responses = []
+                for file_info in sorted_files:
+                    try:
+                        file_storage_id = file_info.get("storage_id")
+                        file_filename = file_info.get("filename")
+                        if not file_storage_id or not file_filename:
+                            raise ValueError(
+                                f"Missing storage_id or filename in manifest file info: {file_info}"
+                            )
+
+                        # Get presigned download URL for this asset
+                        asset_url_response = brain_client.get_presigned_download_url(
+                            file_storage_id, file_filename, minio_endpoint
+                        )
+                        asset_url = asset_url_response["download_url"]
+
+                        # Download asset directly from MinIO
+                        asset_response = requests.get(asset_url, timeout=300)
+                        asset_response.raise_for_status()
+                        asset_responses.append(asset_response.content)
+                    except Exception as e:
+                        logging.error(
+                            f"🛑\u2009 Nilor-Nodes: Failed to download asset {file_info.get('filename')}: {e}"
+                        )
+                        raise  # Re-raise to fail the entire process
+
+                return self._process_image_batch(asset_responses)
+
+            # --- Single-file download ---
+            # Get presigned download URL
+            download_url_response = brain_client.get_presigned_download_url(
+                storage_id, filename, minio_endpoint
+            )
+            download_url = download_url_response["download_url"]
+
+            # Download file directly from MinIO
+            media_response = requests.get(download_url, timeout=300)
+            media_response.raise_for_status()
+            media_bytes = media_response.content
+
+            if format == "video":
+                return self._process_video(media_bytes)
+            elif format == "image":
+                return self._process_image(media_bytes)
+            else:
+                # Should not happen if UI choices are respected
+                raise ValueError(
+                    f"[🛑] Nilor-Nodes (MediaStreamInput): Unsupported format '{format}' for single media download."
+                )
+
+        except Exception as e:
+            logging.error(
+                f"🛑\u2009 Nilor-Nodes (MediaStreamInput): Failed to download or process media: {e}"
+            )
+            return (None,)
+
+    def _process_image_batch(self, image_bytes_list):
+        logging.info(
+            f"ℹ️\u2009 Nilor-Nodes (MediaStreamInput): Processing image batch with {len(image_bytes_list)} images..."
+        )
+        output_images = []
+
+        for image_bytes in image_bytes_list:
+            image_pil = Image.open(io.BytesIO(image_bytes))
+
+            rgb_image_pil = image_pil.convert("RGB")
+            image_tensor = torch.from_numpy(
+                np.array(rgb_image_pil).astype(np.float32) / 255.0
+            ).unsqueeze(0)
+
+            output_images.append(image_tensor)
+
+        # Concatenate along the batch dimension (dim=0)
+        images_tensor = torch.cat(output_images, dim=0)
+
+        logging.info(
+            f"✅ Nilor-Nodes (MediaStreamInput): Image batch processing successful. Batch shape: {images_tensor.shape}"
+        )
+        return (images_tensor,)
+
+    def _process_image(self, image_bytes):
+        logging.info("ℹ️\u2009 Nilor-Nodes (MediaStreamInput): Processing as image...")
+        image_pil = Image.open(io.BytesIO(image_bytes))
+
+        # Ensure image is in RGB
+        rgb_image_pil = image_pil.convert("RGB")
+        image_tensor = torch.from_numpy(
+            np.array(rgb_image_pil).astype(np.float32) / 255.0
+        ).unsqueeze(0)
+
+        logging.info("✅ Nilor-Nodes (MediaStreamInput): Image processing successful.")
+        return (image_tensor,)
+
+    def _process_video(self, video_bytes):
+        logging.info("ℹ️\u2009 Nilor-Nodes (MediaStreamInput): Processing as video...")
+        frames = []
+        with imageio.get_reader(io.BytesIO(video_bytes), format="mp4") as reader:
+            for frame in reader:
+                # Convert frame to RGB PIL Image and then to tensor
+                pil_image = Image.fromarray(frame).convert("RGB")
+                numpy_image = np.array(pil_image).astype(np.float32) / 255.0
+                tensor_frame = torch.from_numpy(numpy_image)
+                frames.append(tensor_frame)
+
+        if not frames:
+            raise ValueError(
+                "[🛑] Nilor-Nodes (MediaStreamInput): No frames could be read from the video."
+            )
+
+        # Stack frames into a single tensor (batch of images)
+        video_tensor = torch.stack(frames)
+
+        logging.info(
+            f"✅ Nilor-Nodes (MediaStreamInput): Video processing successful. Image Shape: {video_tensor.shape}"
+        )
+        return (video_tensor,)
+
+
+# --- MediaStreamOutput: Universal Media Uploader & SQS Notifier ---
+class MediaStreamOutput:
+    @classmethod
+    def INPUT_TYPES(s):
+        return {
+            "required": {
+                "output_name": (
+                    "STRING",
+                    {"default": "default_output", "multiline": False},
+                ),
+                "images": ("IMAGE",),
+                "format": (["png", "mp4"],),
+                "framerate": ("INT", {"default": 24, "min": 1, "max": 240, "step": 1}),
+                "content_id": (
+                    "STRING",
+                    {"default": "<auto-filled by system>", "multiline": False},
+                ),
+                "venue": (
+                    "STRING",
+                    {"default": "<auto-filled by system>", "multiline": False},
+                ),
+                "canvas": (
+                    "STRING",
+                    {"default": "<auto-filled by system>", "multiline": False},
+                ),
+                "scene": (
+                    "STRING",
+                    {"default": "<auto-filled by system>", "multiline": False},
+                ),
+                "job_completions_queue_url": (
+                    "STRING",
+                    {"multiline": True, "default": "<auto-filled by system>"},
+                ),
+            },
+            "hidden": {
+                "prompt": "PROMPT",
+                "extra_pnginfo": "EXTRA_PNGINFO",
+            },
+        }
+
+    RETURN_TYPES = ()
+    FUNCTION = "upload_and_notify"
+    OUTPUT_NODE = True
+    CATEGORY = category + subcategories["streaming"]
+
+    def upload_and_notify(
+        self,
+        images,
+        format,
+        content_id,
+        venue,
+        canvas,
+        scene,
+        job_completions_queue_url,
+        framerate,
+        output_name: str = "default_output",
+        prompt=None,
+        extra_pnginfo=None,
+    ):
+        if not content_id:
+            raise ValueError(
+                "[🛑] Nilor-Nodes (MediaStreamOutput): content_id is a required input for MediaStreamOutput."
+            )
+
+        # No longer need to parse output_object_keys since we use storage_ids directly
+
+        # Upload the media using Brain API client
+        brain_client = get_brain_api_client()
+        storage_result = None
+
+        if format == "png":
+            storage_result = self._upload_image(images[0], brain_client, output_name)
+        elif format == "mp4":
+            storage_result = self._upload_video(
+                images, brain_client, framerate, output_name
+            )
+
+        # Use the storage_id from the upload result for the SQS message
+        if not storage_result:
+            logging.error(
+                f"🛑\u2009 Nilor-Nodes (MediaStreamOutput): FATAL -- Upload failed or no storage_id returned."
+            )
+            # Send an empty dictionary to signal failure.
+            final_outputs_for_sqs = {}
+        else:
+            # Use storage_id directly (it's now a string, not a dict)
+            storage_id = storage_result
+            final_outputs_for_sqs = {output_name: storage_id}
+
+        # After upload, send the filtered dictionary of outputs to the SQS queue.
+        completion_message = {
+            "content_id": content_id,
+            "status": "completed",
+            "venue": venue,
+            "canvas": canvas,
+            "scene": scene,
+            "outputs": final_outputs_for_sqs,
+        }
+
+        try:
+            # Re-initialize the client inside the execution to ensure it picks up env vars correctly.
+            sqs_client = boto3.client(
+                "sqs",
+                endpoint_url=os.getenv("SQS_ENDPOINT_URL"),
+                aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID", "local"),
+                aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY", "local"),
+                region_name=os.getenv("AWS_DEFAULT_REGION", "us-east-1"),
+            )
+            logging.info(
+                f"ℹ️\u2009 Nilor-Nodes (MediaStreamOutput): Sending completion message for content {content_id} to queue: {job_completions_queue_url}"
+            )
+            sqs_client.send_message(
+                QueueUrl=job_completions_queue_url,
+                MessageBody=json.dumps(completion_message),
+            )
+            logging.info(
+                "✅ Nilor-Nodes (MediaStreamOutput): Completion message sent successfully."
+            )
+        except Exception as e:
+            logging.error(
+                f"🛑\u2009 Nilor-Nodes (MediaStreamOutput): Failed to send completion message to SQS: {e}"
+            )
+            raise  # Re-raise to fail the ComfyUI job
+
+        return {"ui": {"images": []}}
+
+    def _upload_image(self, image_tensor, brain_client, output_name):
+        logging.info(
+            "ℹ️\u2009 Nilor-Nodes (MediaStreamOutput): Uploading as PNG image..."
+        )
+        i = 255.0 * image_tensor.cpu().numpy()
+        img_pil = Image.fromarray(np.clip(i, 0, 255).astype(np.uint8))
+
+        buffer = io.BytesIO()
+        img_pil.save(buffer, format="PNG", compress_level=4)
+        buffer.seek(0)
+
+        filename = f"{output_name}.png"
+        minio_endpoint = os.getenv("MINIO_ENDPOINT")
+        if not minio_endpoint:
+            raise ValueError(
+                "MINIO_ENDPOINT environment variable is required but not set"
+            )
+
+        # Get presigned upload URL
+        upload_url_response = brain_client.get_presigned_upload_url(
+            filename, "image/png", minio_endpoint
+        )
+        upload_url = upload_url_response["upload_url"]
+        storage_id = upload_url_response["storage_id"]
+
+        # Upload directly to MinIO
+        buffer.seek(0)
+        upload_response = requests.put(
+            upload_url,
+            data=buffer.getvalue(),
+            headers={"Content-Type": "image/png"},
+            timeout=300,
+        )
+        upload_response.raise_for_status()
+
+        logging.info(
+            f"✅ Nilor-Nodes (MediaStreamOutput): PNG image uploaded successfully. Storage ID: {storage_id}"
+        )
+        return storage_id
+
+    def _upload_video(self, image_batch_tensor, brain_client, framerate, output_name):
+        logging.info(
+            f"ℹ️\u2009 Nilor-Nodes (MediaStreamOutput): Uploading as MP4 video. Frame count: {len(image_batch_tensor)}"
+        )
+        frames = []
+        for image_tensor in image_batch_tensor:
+            i = 255.0 * image_tensor.cpu().numpy()
+            frame = np.clip(i, 0, 255).astype(np.uint8)
+            frames.append(frame)
+
+        buffer = io.BytesIO()
+        imageio.mimwrite(buffer, frames, format="mp4", fps=framerate, quality=8)
+        buffer.seek(0)
+
+        filename = f"{output_name}.mp4"
+        minio_endpoint = os.getenv("MINIO_ENDPOINT")
+        if not minio_endpoint:
+            raise ValueError(
+                "MINIO_ENDPOINT environment variable is required but not set"
+            )
+
+        # Get presigned upload URL
+        upload_url_response = brain_client.get_presigned_upload_url(
+            filename, "video/mp4", minio_endpoint
+        )
+        upload_url = upload_url_response["upload_url"]
+        storage_id = upload_url_response["storage_id"]
+
+        # Upload directly to MinIO
+        buffer.seek(0)
+        upload_response = requests.put(
+            upload_url,
+            data=buffer.getvalue(),
+            headers={"Content-Type": "video/mp4"},
+            timeout=300,
+        )
+        upload_response.raise_for_status()
+
+        logging.info(
+            f"✅ Nilor-Nodes (MediaStreamOutput): MP4 video uploaded successfully. Storage ID: {storage_id}"
+        )
+        return storage_id
+
+
+# --- Node Mappings ---
+NODE_CLASS_MAPPINGS = {
+    "MediaStreamInput": MediaStreamInput,
+    "MediaStreamOutput": MediaStreamOutput,
+}
+
+NODE_DISPLAY_NAME_MAPPINGS = {
+    "MediaStreamInput": "👺 Media Stream Input (Storage)",
+    "MediaStreamOutput": "👺 Media Stream Output (Storage)",
+}

nilornodes.py

@@ -16,7 +16,24 @@ import torch
 import builtins
 from pathlib import Path
 import cv2
+import warnings
 from .utils import pil2tensor, tensor2pil
+import logging
+from comfy.utils import common_upscale
+from comfy import model_management
+import sys
+from os.path import dirname, join
+
+# Attempt to import ImagePadKJ from comfyui-kjnodes if available
+_kj_nodes_path = join(dirname(__file__), "..", "comfyui-kjnodes", "nodes")
+if _kj_nodes_path not in sys.path:
+    sys.path.append(_kj_nodes_path)
+try:
+    from image_nodes import ImagePadKJ  # type: ignore
+except Exception as _e:
+    logging.warning(
+        f"⚠️\u2009 Nilor-Nodes (nilornodes): Could not import ImagePadKJ from comfyui-kjnodes ({_kj_nodes_path}): {_e}"
+    )
 
 BIGMIN = -(2**53 - 1)
 BIGMAX = 2**53 - 1
@@ -166,7 +183,9 @@ class NilorRemapFloatList:
     ):
         # Avoid division by zero
         if max_input - min_input == 0:
-            raise ValueError("max_input and min_input cannot be the same value.")
+            raise ValueError(
+                "🛑\u2009 Nilor-Nodes (RemapFloatList): max_input and min_input cannot be the same value."
+            )
 
         scale = (max_output - min_output) / (max_input - min_input)
         return ([min_output + (x - min_input) * scale for x in list_of_floats],)
@@ -221,7 +240,9 @@ class NilorInverseMapFloatList:
 
     def inverse_map_float_list(self, list_of_floats):
         if not list_of_floats:
-            raise ValueError("The input list_of_floats cannot be empty.")
+            raise ValueError(
+                "🛑\u2009 Nilor-Nodes (InverseMapFloatList): The input list_of_floats cannot be empty."
+            )
 
         min_input = min(list_of_floats)
         max_input = max(list_of_floats)
@@ -317,7 +338,9 @@ class NilorCountImagesInDirectory:
 
     def count_images_in_directory(self, directory):
         if not os.path.isdir(directory):
-            raise FileNotFoundError(f"Directory '{directory} cannot be found.")
+            raise FileNotFoundError(
+                f"🛑\u2009 Nilor-Nodes (NilorCountImagesInDirectory): Directory '{directory}' cannot be found."
+            )
 
         list_dir = []
         list_dir = os.listdir(directory)
@@ -365,7 +388,9 @@ class NilorSelectIndexFromList:
 
         # Ensure the index is within bounds
         if index < 0 or index >= len(actual_list):
-            raise ValueError("Index is outside the bounds of the array.")
+            raise ValueError(
+                "🛑\u2009 Nilor-Nodes (SelectIndexFromList): Index is outside the bounds of the array."
+            )
 
         # Returns the value at the given index
         return (actual_list[index],)
@@ -403,7 +428,9 @@ class NilorSaveEXRArbitrary:
         self, channels=None, filename_prefix="output", prompt=None, extra_pnginfo=None
     ):
 
-        print("Running save_exr_arbitrary")
+        logging.info(
+            "ℹ️\u2009 Nilor-Nodes (SaveEXRArbitrary): Running save_exr_arbitrary"
+        )
         # print(f"channels: {channels}")
         # print(f"filename_prefix: {filename_prefix}")
 
@@ -415,7 +442,9 @@ class NilorSaveEXRArbitrary:
         try:
             actual_channels[0]
         except TypeError:
-            print("actual_channels is not subscriptable")
+            logging.error(
+                "🛑\u2009 Nilor-Nodes (SaveEXRArbitrary): actual_channels is not subscriptable"
+            )
             return
 
         # File path handling
@@ -452,7 +481,9 @@ class NilorSaveEXRArbitrary:
             height, width = image_channels[0].shape[-2:]
             for tensor in image_channels:
                 if tensor.shape[-2:] != (height, width):
-                    raise ValueError("All input tensors must have the same dimensions")
+                    raise ValueError(
+                        "🛑\u2009 Nilor-Nodes (SaveEXRArbitrary): All input tensors must have the same dimensions"
+                    )
 
             # Channel naming
             default_names = ["R", "G", "B", "A"] + [
@@ -502,9 +533,13 @@ class NilorSaveEXRArbitrary:
             exr_file.writePixels(channel_data)
             exr_file.close()
 
-            print(f"EXR file saved successfully to {writepath}")
+            logging.info(
+                f"✅\u2009 Nilor-Nodes (SaveEXRArbitrary): EXR file saved successfully to {writepath}"
+            )
         except Exception as e:
-            print(f"Failed to write EXR file: {e}")
+            logging.error(
+                f"🛑\u2009 Nilor-Nodes (SaveEXRArbitrary): Failed to write EXR file: {e}"
+            )
 
 
 class NilorSaveVideoToHFDataset:
@@ -622,12 +657,14 @@ class NilorShuffleImageBatch:
 
     def _check_image_dimensions(self, images):
         if images.shape[0] == 0:
-            raise ValueError("Input images tensor is empty.")
+            raise ValueError(
+                "🛑\u2009 Nilor-Nodes (ShuffleImageBatch): Input images tensor is empty."
+            )
 
         # All images in the batch should have the same dimensions
         if len(images.shape) != 4:
             raise ValueError(
-                f"Expected 4D tensor (batch, channels, height, width), got shape {images.shape}"
+                f"🛑\u2009 Nilor-Nodes (ShuffleImageBatch): Expected 4D tensor (batch, channels, height, width), got shape {images.shape}"
             )
 
     def shuffle_image_batch(self, images: torch.Tensor, seed):
@@ -667,12 +704,14 @@ class NilorRepeatTrimImageBatch:
 
     def _check_image_dimensions(self, images):
         if images.shape[0] == 0:
-            raise ValueError("Input images tensor is empty.")
+            raise ValueError(
+                "🛑\u2009 Nilor-Nodes (RepeatTrimImageBatch): Input images tensor is empty."
+            )
 
         # All images in the batch should have the same dimensions
         if len(images.shape) != 4:
             raise ValueError(
-                f"Expected 4D tensor (batch, channels, height, width), got shape {images.shape}"
+                f"🛑\u2009 Nilor-Nodes (RepeatTrimImageBatch): Expected 4D tensor (batch, channels, height, width), got shape {images.shape}"
             )
 
     def repeat_trim_image_batch(self, images: torch.Tensor, count):
@@ -710,12 +749,14 @@ class NilorRepeatShuffleTrimImageBatch:
 
     def _check_image_dimensions(self, images):
         if images.shape[0] == 0:
-            raise ValueError("Input images tensor is empty.")
+            raise ValueError(
+                "🛑\u2009 Nilor-Nodes (RepeatShuffleTrimImageBatch): Input images tensor is empty."
+            )
 
         # All images in the batch should have the same dimensions
         if len(images.shape) != 4:
             raise ValueError(
-                f"Expected 4D tensor (batch, channels, height, width), got shape {images.shape}"
+                f"🛑\u2009 Nilor-Nodes (RepeatShuffleTrimImageBatch): Expected 4D tensor (batch, channels, height, width), got shape {images.shape}"
             )
 
     def repeat_shuffle_trim_image_batch(self, images: torch.Tensor, seed, count):
@@ -781,12 +822,16 @@ class NilorOutputFilenameString:
 
         if unique_id is not None and extra_pnginfo is not None:
             if not isinstance(extra_pnginfo, list):
-                print("Error: extra_pnginfo is not a list")
+                logging.error(
+                    "🛑\u2009 Nilor-Nodes (OutputFilenameString): extra_pnginfo is not a list"
+                )
             elif (
                 not isinstance(extra_pnginfo[0], dict)
                 or "workflow" not in extra_pnginfo[0]
             ):
-                print("Error: extra_pnginfo[0] is not a dict or missing 'workflow' key")
+                logging.error(
+                    "🛑\u2009 Nilor-Nodes (OutputFilenameString): extra_pnginfo[0] is not a dict or missing 'workflow' key"
+                )
             else:
                 workflow = extra_pnginfo[0]["workflow"]
                 node = next(
@@ -837,7 +882,206 @@ class NilorNFractionsOfInt:
         elif type == "start + end":
             return ([i * numerator // (denominator - 1) for i in range(denominator)],)
         else:
-            raise ValueError(f"Unknown type: {type}")
+            raise ValueError(
+                f"🛑\u2009 Nilor-Nodes (NilorNFractionsOfInt): Unknown type: {type}"
+            )
+
+
+class NilorWanTileResolution:
+    def __init__(self):
+        pass
+
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "input_width": (
+                    "INT",
+                    {"default": 1920, "min": 16, "max": BIGMAX, "step": 1},
+                ),
+                "input_height": (
+                    "INT",
+                    {"default": 1080, "min": 16, "max": BIGMAX, "step": 1},
+                ),
+                "target_width": (
+                    "INT",
+                    {"default": 3840, "min": 16, "max": BIGMAX, "step": 1},
+                ),
+                "target_height": (
+                    "INT",
+                    {"default": 2160, "min": 16, "max": BIGMAX, "step": 1},
+                ),
+                "size_preference": (
+                    ["largest", "smallest"],
+                    {"default": "largest"},
+                ),
+            }
+        }
+
+    RETURN_TYPES = ("INT", "INT")
+    RETURN_NAMES = ("tile_width", "tile_height")
+
+    FUNCTION = "compute_tile_resolution"
+    CATEGORY = category + subcategories["utilities"]
+
+    MIN_TILE_DIM = 384
+    MAX_TILE_DIM = 1794
+    MIN_TILE_AREA = 384 * 384
+    MAX_TILE_AREA = 1024 * 1024
+
+    @staticmethod
+    def _clamp(value, minimum, maximum):
+        return max(minimum, min(value, maximum))
+
+    def compute_tile_resolution(
+        self,
+        input_width,
+        input_height,
+        target_width,
+        target_height,
+        size_preference="largest",
+    ):
+        """
+        Compute (Wt, Ht) tile size (multiples of 16) within
+        [MIN_TILE_DIM, MAX_TILE_DIM] while keeping area between
+        [MIN_TILE_AREA, MAX_TILE_AREA]. Emphasise aspect-ratio fidelity to
+        Wa/Ha while staying within the allowed range.
+
+        Among options with comparable aspect error, prefer tiles that do
+        not hit clamped bounds, then maximise area and width (or minimise both if
+        size_preference == "smallest").
+
+        Assumes Wa, Ha are multiples of 16.
+        """
+
+        dims = {
+            "input_width": input_width,
+            "input_height": input_height,
+            "target_width": target_width,
+            "target_height": target_height,
+        }
+
+        for name, value in dims.items():
+            if value <= 0:
+                raise ValueError(
+                    f"🛑\u2009 Nilor-Nodes (NilorWanTileResolution): {name} must be a positive integer."
+                )
+
+        if input_width % 16 != 0 or input_height % 16 != 0:
+            raise ValueError(
+                "🛑\u2009 Nilor-Nodes (NilorWanTileResolution): input_width and input_height must be multiples of 16."
+            )
+
+        if target_width < self.MIN_TILE_DIM or target_height < self.MIN_TILE_DIM:
+            raise ValueError(
+                "🛑\u2009 Nilor-Nodes (NilorWanTileResolution): target_width and target_height must be at least the minimum tile size."
+            )
+
+        min_blocks = self.MIN_TILE_DIM // 16
+        max_blocks = self.MAX_TILE_DIM // 16
+
+        max_width_blocks = min(max_blocks, target_width // 16)
+        max_height_blocks = min(max_blocks, target_height // 16)
+
+        if max_width_blocks < min_blocks or max_height_blocks < min_blocks:
+            raise ValueError(
+                "🛑\u2009 Nilor-Nodes (NilorWanTileResolution): Target dimensions do not allow a tile within the supported range."
+            )
+
+        aspect_ratio = input_width / input_height
+
+        best_score = None
+        best_dimensions = None
+
+        for height_blocks in range(min_blocks, max_height_blocks + 1):
+            width_blocks = round(aspect_ratio * height_blocks)
+            width_blocks = self._clamp(width_blocks, min_blocks, max_width_blocks)
+
+            width_px = width_blocks * 16
+            height_px = height_blocks * 16
+
+            area = width_px * height_px
+            if area < self.MIN_TILE_AREA or area > self.MAX_TILE_AREA:
+                # Skip tiles that are too small or too large
+                continue
+            aspect_error = abs((width_blocks / height_blocks) - aspect_ratio)
+
+            width_hits_bound = int(width_blocks in (min_blocks, max_width_blocks))
+            height_hits_bound = int(height_blocks in (min_blocks, max_height_blocks))
+
+            # Penalise tiles that hit the clamped bounds
+            bound_penalty = width_hits_bound + height_hits_bound
+
+            # Score tiles based on size preference
+            if size_preference == "smallest":
+                area_score = -area
+                width_score = -width_px
+            else:
+                area_score = area
+                width_score = width_px
+
+            # Combine scores
+            candidate = (-aspect_error, -bound_penalty, area_score, width_score)
+
+            if best_score is None or candidate > best_score:
+                # Update best score and dimensions if this candidate is better
+                best_score = candidate
+                best_dimensions = (width_px, height_px)
+
+        if best_dimensions is None:
+            # If no suitable tile resolution was found, raise an error
+            raise RuntimeError(
+                "🛑\u2009 Nilor-Nodes (NilorWanTileResolution): Failed to determine a suitable tile resolution."
+            )
+
+        return best_dimensions
+
+
+class NilorWanFrameTrim:
+    def __init__(self):
+        pass
+
+    @classmethod
+    def INPUT_TYPES(s):
+        return {
+            "required": {
+                "images": ("IMAGE",),
+            },
+        }
+
+    RETURN_TYPES = ("IMAGE",)
+    RETURN_NAMES = ("images",)
+    FUNCTION = "trim_to_wan_count"
+    CATEGORY = category + subcategories["utilities"]
+
+    def _validate_images(self, images):
+        if not isinstance(images, torch.Tensor):
+            raise TypeError(
+                "🛑\u2009 Nilor-Nodes (WanFrameTrim): images must be a torch.Tensor."
+            )
+        if images.dim() != 4:
+            raise ValueError(
+                f"🛑\u2009 Nilor-Nodes (WanFrameTrim): Expected 4D tensor (batch, height, width, channels), got shape {tuple(images.shape)}"
+            )
+        if images.shape[0] == 0:
+            raise ValueError(
+                "🛑\u2009 Nilor-Nodes (WanFrameTrim): Input images tensor is empty."
+            )
+
+    def trim_to_wan_count(self, images: torch.Tensor):
+        self._validate_images(images)
+
+        batch_count = images.shape[0]
+        # Find the largest m <= batch_count such that m ≡ 1 (mod 4)
+        wan_count = batch_count - ((batch_count - 1) % 4)
+
+        if wan_count <= 0:
+            raise ValueError(
+                "🛑\u2009 Nilor-Nodes (WanFrameTrim): Unable to compute a valid 4N+1 frame count from input."
+            )
+
+        trimmed = images[:wan_count]
+        return (trimmed,)
 
 
 class NilorCategorizeString:
@@ -959,7 +1203,9 @@ class NilorRandomString:
             if item.strip()
         ]
         if not options:
-            raise ValueError("No valid choices provided.")
+            raise ValueError(
+                "🛑\u2009 Nilor-Nodes (NilorRandomString): No valid choices provided."
+            )
 
         # Limit to the first 'max_options' entries if there are more options
         if len(options) > max_options:
@@ -1001,7 +1247,9 @@ class NilorLoadImageByIndex:
 
     def load_image_by_index(self, image_directory, seed, sort_mode, reverse_sort):
         if not os.path.exists(image_directory):
-            raise FileNotFoundError(f"Image directory {image_directory} does not exist")
+            raise FileNotFoundError(
+                f"🛑\u2009 Nilor-Nodes (NilorLoadImageByIndex): Image directory {image_directory} does not exist"
+            )
 
         # Get list of image files
         files = []
@@ -1013,7 +1261,9 @@ class NilorLoadImageByIndex:
                 files.append(file_path)
 
         if not files:
-            raise ValueError(f"No image files found in {image_directory}")
+            raise ValueError(
+                f"🛑\u2009 Nilor-Nodes (NilorLoadImageByIndex): No image files found in {image_directory}"
+            )
 
         # Sort files based on selected mode
         if sort_mode == "filename":
@@ -1063,7 +1313,9 @@ class NilorExtractFilenameFromPath:
     def extract_filename(self, filepath):
         # Ensure the input is a valid path
         if not filepath:
-            raise ValueError("Filepath cannot be empty.")
+            raise ValueError(
+                "🛑\u2009 Nilor-Nodes (ExtractFilenameFromPath): Filepath cannot be empty."
+            )
 
         path = Path(filepath)
 
@@ -1098,14 +1350,18 @@ class NilorBlurAnalysis:
         """
         # Ensure images is a 4D tensor.
         if images.dim() != 4:
-            raise ValueError("Input images must be a 4D tensor (batch, channels/height, height/width, width/channels)")
+            raise ValueError(
+                "🛑\u2009 Nilor-Nodes (BlurAnalysis): Input images must be a 4D tensor (batch, channels/height, height/width, width/channels)"
+            )
 
         # Detect if using NCHW or NHWC.
         if images.shape[1] not in (1, 3):
             if images.shape[-1] in (1, 3):
                 images = images.permute(0, 3, 1, 2)
             else:
-                raise ValueError("Cannot determine image format (expected channel to be 1 or 3).")
+                raise ValueError(
+                    "🛑\u2009 Nilor-Nodes (BlurAnalysis): Cannot determine image format (expected channel to be 1 or 3)."
+                )
 
         output_images = []
         batch_size = images.shape[0]
@@ -1116,9 +1372,7 @@ class NilorBlurAnalysis:
 
             # Convert to grayscale.
             if img_np.shape[0] >= 3:
-                gray = (0.299 * img_np[0] +
-                        0.587 * img_np[1] +
-                        0.114 * img_np[2])
+                gray = 0.299 * img_np[0] + 0.587 * img_np[1] + 0.114 * img_np[2]
             else:
                 gray = np.squeeze(img_np, axis=0)  # shape: (H, W)
 
@@ -1146,7 +1400,9 @@ class NilorBlurAnalysis:
             # Convert the single channel output to a 3-channel image.
             # This ensures downstream nodes (like MaskFromRGBCMYBW) that index into channels work properly.
             if out_img.ndim == 2:
-                out_img = np.stack([out_img, out_img, out_img], axis=-1)  # shape becomes (H, W, 3)
+                out_img = np.stack(
+                    [out_img, out_img, out_img], axis=-1
+                )  # shape becomes (H, W, 3)
 
             # Convert from PIL image (or numpy array) to tensor.
             # pil2tensor should create a tensor in a format that downstream nodes expect.
@@ -1157,6 +1413,7 @@ class NilorBlurAnalysis:
         # If each output has shape, say, (H, W, 3), stacking them gives a tensor of shape (B, H, W, 3).
         return (torch.cat(output_images, dim=0),)
 
+
 class NilorToSparseIndexMethod:
     def __init__(self):
         pass
@@ -1179,10 +1436,315 @@ class NilorToSparseIndexMethod:
 
     def convert_to_sparse_index_method(self, ints):
         indexes_str = ",".join(map(str, ints))
-        
+
         return (indexes_str,)
 
 
+class NilorImageResizeV2:
+    upscale_methods = ["nearest-exact", "bilinear", "area", "bicubic", "lanczos"]
+
+    @classmethod
+    def INPUT_TYPES(s):
+        return {
+            "required": {
+                "image": ("IMAGE",),
+                "width": ("INT", {"default": 512, "min": 0, "max": BIGMAX, "step": 1}),
+                "height": ("INT", {"default": 512, "min": 0, "max": BIGMAX, "step": 1}),
+                "upscale_method": (s.upscale_methods,),
+                "keep_proportion": (
+                    [
+                        "stretch",
+                        "resize",
+                        "pad",
+                        "pad_edge",
+                        "pad_edge_pixel",
+                        "crop",
+                        "pillarbox_blur",
+                    ],
+                    {"default": False},
+                ),
+                "pad_color": ("STRING", {"default": "0, 0, 0"}),
+                "crop_position": (
+                    ["center", "top", "bottom", "left", "right"],
+                    {"default": "center"},
+                ),
+                "divisible_by": (
+                    "INT",
+                    {"default": 2, "min": 0, "max": 512, "step": 1},
+                ),
+            },
+            "optional": {
+                "mask": ("MASK",),
+                "device": (["cpu", "gpu"],),
+                "per_batch": (
+                    "INT",
+                    {
+                        "default": 16,
+                        "min": 0,
+                        "max": 4096,
+                        "step": 1,
+                        "tooltip": "Process images in sub-batches. 0 disables.",
+                    },
+                ),
+            },
+            "hidden": {"unique_id": "UNIQUE_ID"},
+        }
+
+    RETURN_TYPES = ("IMAGE", "INT", "INT", "MASK")
+    RETURN_NAMES = ("IMAGE", "width", "height", "mask")
+    FUNCTION = "resize"
+    CATEGORY = category + subcategories["utilities"]
+    DESCRIPTION = """
+Resizes images with optional aspect preservation, padding/cropping, and sub-batching to lower peak memory.
+"""
+
+    def resize(
+        self,
+        image,
+        width,
+        height,
+        keep_proportion,
+        upscale_method,
+        divisible_by,
+        pad_color,
+        crop_position,
+        unique_id,
+        device="cpu",
+        mask=None,
+        per_batch=16,
+    ):
+        B, H, W, C = image.shape
+
+        if device == "gpu":
+            if upscale_method == "lanczos":
+                raise Exception(
+                    "🛑\u2009 Nilor-Nodes (NilorImageResizeV2): Lanczos is not supported on the GPU"
+                )
+            device = model_management.get_torch_device()
+        else:
+            device = torch.device("cpu")
+
+        if width == 0:
+            width = W
+        if height == 0:
+            height = H
+
+        pillarbox_blur = keep_proportion == "pillarbox_blur"
+        if (
+            keep_proportion == "resize"
+            or keep_proportion.startswith("pad")
+            or pillarbox_blur
+        ):
+            if width == 0 and height != 0:
+                ratio = height / H
+                new_width = round(W * ratio)
+                new_height = height
+            elif height == 0 and width != 0:
+                ratio = width / W
+                new_width = width
+                new_height = round(H * ratio)
+            elif width != 0 and height != 0:
+                ratio = min(width / W, height / H)
+                new_width = round(W * ratio)
+                new_height = round(H * ratio)
+            else:
+                new_width = width
+                new_height = height
+
+            pad_left = pad_right = pad_top = pad_bottom = 0
+            if keep_proportion.startswith("pad") or pillarbox_blur:
+                if crop_position == "center":
+                    pad_left = (width - new_width) // 2
+                    pad_right = width - new_width - pad_left
+                    pad_top = (height - new_height) // 2
+                    pad_bottom = height - new_height - pad_top
+                elif crop_position == "top":
+                    pad_left = (width - new_width) // 2
+                    pad_right = width - new_width - pad_left
+                    pad_top = 0
+                    pad_bottom = height - new_height
+                elif crop_position == "bottom":
+                    pad_left = (width - new_width) // 2
+                    pad_right = width - new_width - pad_left
+                    pad_top = height - new_height
+                    pad_bottom = 0
+                elif crop_position == "left":
+                    pad_left = 0
+                    pad_right = width - new_width
+                    pad_top = (height - new_height) // 2
+                    pad_bottom = height - new_height - pad_top
+                elif crop_position == "right":
+                    pad_left = width - new_width
+                    pad_right = 0
+                    pad_top = (height - new_height) // 2
+                    pad_bottom = height - new_height - pad_top
+
+            width = new_width
+            height = new_height
+
+        if divisible_by > 1:
+            width = width - (width % divisible_by)
+            height = height - (height % divisible_by)
+
+        if per_batch and B > per_batch:
+            try:
+                bytes_per_elem = image.element_size()
+                est_total_bytes = B * height * width * C * bytes_per_elem
+                est_mb = est_total_bytes / (1024 * 1024)
+                logging.info(
+                    f"ℹ️\u2009 Nilor-Nodes (NilorImageResizeV2) Estimated output ~{est_mb:.2f} MB."
+                )
+            except:
+                pass
+
+        def _process_subbatch(in_image, in_mask):
+            out_image = in_image if in_image.device == device else in_image.to(device)
+            out_mask = (
+                None
+                if in_mask is None
+                else (in_mask if in_mask.device == device else in_mask.to(device))
+            )
+
+            if keep_proportion == "crop":
+                old_height = out_image.shape[-3]
+                old_width = out_image.shape[-2]
+                old_aspect = old_width / old_height
+                new_aspect = width / height
+                if old_aspect > new_aspect:
+                    crop_w = round(old_height * new_aspect)
+                    crop_h = old_height
+                else:
+                    crop_w = old_width
+                    crop_h = round(old_width / new_aspect)
+                if crop_position == "center":
+                    x = (old_width - crop_w) // 2
+                    y = (old_height - crop_h) // 2
+                elif crop_position == "top":
+                    x = (old_width - crop_w) // 2
+                    y = 0
+                elif crop_position == "bottom":
+                    x = (old_width - crop_w) // 2
+                    y = old_height - crop_h
+                elif crop_position == "left":
+                    x = 0
+                    y = (old_height - crop_h) // 2
+                elif crop_position == "right":
+                    x = old_width - crop_w
+                    y = (old_height - crop_h) // 2
+                out_image = out_image.narrow(-2, x, crop_w).narrow(-3, y, crop_h)
+                if out_mask is not None:
+                    out_mask = out_mask.narrow(-1, x, crop_w).narrow(-2, y, crop_h)
+
+            out_image = common_upscale(
+                out_image.movedim(-1, 1), width, height, upscale_method, crop="disabled"
+            ).movedim(1, -1)
+            if out_mask is not None:
+                if upscale_method == "lanczos":
+                    out_mask = common_upscale(
+                        out_mask.unsqueeze(1).repeat(1, 3, 1, 1),
+                        width,
+                        height,
+                        upscale_method,
+                        crop="disabled",
+                    ).movedim(1, -1)[:, :, :, 0]
+                else:
+                    out_mask = common_upscale(
+                        out_mask.unsqueeze(1),
+                        width,
+                        height,
+                        upscale_method,
+                        crop="disabled",
+                    ).squeeze(1)
+
+            if (keep_proportion.startswith("pad") or pillarbox_blur) and (
+                pad_left > 0 or pad_right > 0 or pad_top > 0 or pad_bottom > 0
+            ):
+                padded_width = width + pad_left + pad_right
+                padded_height = height + pad_top + pad_bottom
+                if divisible_by > 1:
+                    width_remainder = padded_width % divisible_by
+                    height_remainder = padded_height % divisible_by
+                    if width_remainder > 0:
+                        extra_width = divisible_by - width_remainder
+                        pad_right += extra_width
+                    if height_remainder > 0:
+                        extra_height = divisible_by - height_remainder
+                        pad_bottom += extra_height
+
+                pad_mode = (
+                    "pillarbox_blur"
+                    if pillarbox_blur
+                    else (
+                        "edge"
+                        if keep_proportion == "pad_edge"
+                        else (
+                            "edge_pixel"
+                            if keep_proportion == "pad_edge_pixel"
+                            else "color"
+                        )
+                    )
+                )
+                out_image, out_mask = ImagePadKJ.pad(
+                    self,
+                    out_image,
+                    pad_left,
+                    pad_right,
+                    pad_top,
+                    pad_bottom,
+                    0,
+                    pad_color,
+                    pad_mode,
+                    mask=out_mask,
+                )
+
+            return out_image, out_mask
+
+        if per_batch is None or per_batch == 0 or B <= per_batch:
+            out_image, out_mask = _process_subbatch(image, mask)
+        else:
+            chunks = []
+            mask_chunks = [] if mask is not None else None
+            total_batches = (B + per_batch - 1) // per_batch
+            current_batch = 0
+            for start_idx in range(0, B, per_batch):
+                current_batch += 1
+                end_idx = min(start_idx + per_batch, B)
+                sub_img = image[start_idx:end_idx]
+                sub_mask = mask[start_idx:end_idx] if mask is not None else None
+                sub_out_img, sub_out_mask = _process_subbatch(sub_img, sub_mask)
+                chunks.append(sub_out_img.cpu())
+                if mask is not None:
+                    mask_chunks.append(
+                        sub_out_mask.cpu() if sub_out_mask is not None else None
+                    )
+                try:
+                    logging.info(
+                        f"ℹ️\u2009 Nilor-Nodes (NilorImageResizeV2) Batch {current_batch}/{total_batches} · images {end_idx}/{B}"
+                    )
+                except:
+                    pass
+            out_image = torch.cat(chunks, dim=0)
+            if mask is not None and any(m is not None for m in mask_chunks):
+                out_mask = torch.cat([m for m in mask_chunks if m is not None], dim=0)
+            else:
+                out_mask = None
+
+        logging.info(f"✅\u2009 Nilor-Nodes (NilorImageResizeV2) All batches complete.")
+
+        return (
+            out_image.cpu(),
+            out_image.shape[2],
+            out_image.shape[1],
+            (
+                out_mask.cpu()
+                if out_mask is not None
+                else torch.zeros(
+                    64, 64, device=torch.device("cpu"), dtype=torch.float32
+                )
+            ),
+        )
+
+
 # Mapping class names to objects for potential export
 NODE_CLASS_MAPPINGS = {
     "Nilor Interpolated Float List": NilorInterpolatedFloatList,
@@ -1204,19 +1766,22 @@ NODE_CLASS_MAPPINGS = {
     "Nilor n Fractions of Int": NilorNFractionsOfInt,
     "Nilor Categorize String": NilorCategorizeString,
     "Nilor Random String": NilorRandomString,
+    "Nilor Wan Tile Resolution": NilorWanTileResolution,
     "Nilor Extract Filename from Path": NilorExtractFilenameFromPath,
     "Nilor Load Image By Index": NilorLoadImageByIndex,
     "Nilor Blur Analysis": NilorBlurAnalysis,
     "Nilor To Sparse Index Method": NilorToSparseIndexMethod,
+    "Nilor Image Resize v2": NilorImageResizeV2,
+    "Nilor Wan Frame Trim": NilorWanFrameTrim,
 }
 
 # Mapping nodes to human-readable names
 NODE_DISPLAY_NAME_MAPPINGS = {
     "Nilor Interpolated Float List": "👺 Interpolated Float List",
     "Nilor One Minus Float List": "👺 One Minus Float List",
-    "Nilor Remap Float List": "👺 Nilor Remap Float List",
-    "Nilor Remap Float List Auto Input": "👺 Nilor Remap Float List Auto Input",
-    "Nilor Inverse Map Float List": "👺 Nilor Inverse Map Float List",
+    "Nilor Remap Float List": "👺 Remap Float List",
+    "Nilor Remap Float List Auto Input": "👺 Remap Float List Auto Input",
+    "Nilor Inverse Map Float List": "👺 Inverse Map Float List",
     "Nilor Int To List Of Bools": "👺 Int To List Of Bools",
     "Nilor List of Ints": "👺 List of Ints",
     "Nilor Count Images In Directory": "👺 Count Images In Directory",
@@ -1224,15 +1789,18 @@ NODE_DISPLAY_NAME_MAPPINGS = {
     "Nilor Save Video To HF Dataset": "👺 Save Video To HF Dataset",
     "Nilor Select Index From List": "👺 Select Index From List",
     "Nilor Save EXR Arbitrary": "👺 Save EXR Arbitrary",
-    "Nilor Shuffle Image Batch": "👺 Nilor Shuffle Image Batch",
-    "Nilor Repeat & Trim Image Batch": "👺 Nilor Repeat & Trim Image Batch",
-    "Nilor Repeat, Shuffle, & Trim Image Batch": "👺 Nilor Repeat, Shuffle, & Trim Image Batch",
-    "Nilor Output Filename String": "👺 Nilor Output Filename String",
-    "Nilor n Fractions of Int": "👺 Nilor n Fractions of Int",
+    "Nilor Shuffle Image Batch": "👺 Shuffle Image Batch",
+    "Nilor Repeat & Trim Image Batch": "👺 Repeat & Trim Image Batch",
+    "Nilor Repeat, Shuffle, & Trim Image Batch": "👺 Repeat, Shuffle, & Trim Image Batch",
+    "Nilor Output Filename String": "👺 Output Filename String",
+    "Nilor n Fractions of Int": "👺 n Fractions of Int",
     "Nilor Categorize String": "👺 Categorize String",
     "Nilor Random String": "👺 Random String",
+    "Nilor Wan Tile Resolution": "👺 Wan Tile Resolution",
     "Nilor Extract Filename from Path": "👺 Extract Filename from Path",
     "Nilor Load Image By Index": "👺 Load Image By Index",
     "Nilor Blur Analysis": "👺 Blur Analysis",
     "Nilor To Sparse Index Method": "👺 To Sparse Index Method",
+    "Nilor Image Resize v2": "👺 Resize Image v2",
+    "Nilor Wan Frame Trim": "👺 Wan Frame Trim",
 }

requirements.txt

@@ -1,2 +1,19 @@
-huggingface_hub
-openexr
+
+aiobotocore==2.24.2
+aiofiles>=23.2.1
+aiohttp==3.12.14
+boto3==1.40.15
+fastapi==0.110.0
+huggingface_hub==0.34.0
+imageio==2.37.0
+imageio-ffmpeg==0.6.0
+numpy>=1.26.4
+opencv-python>=4.6.0.66
+openexr==3.3.4
+Pillow==10.4.0
+python-dotenv==1.0.1
+python-multipart==0.0.9
+requests==2.31.0
+uvicorn==0.27.1
+websockets==11.0.3
+--prefer-binary

user_input.py

@@ -0,0 +1,109 @@
+category = "Nilor Nodes 👺"
+subcategories = {
+    "io": "/IO",
+}
+
+from .controllers import CONTROLLER_HOOK
+
+
+class NilorUserInput_String:
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "input_name": (
+                    "STRING",
+                    {"default": "my_string_input", "multiline": False},
+                ),
+                "value": ("STRING", {"default": "", "multiline": True}),
+            }
+        }
+
+    RETURN_TYPES = ("STRING", CONTROLLER_HOOK)
+    RETURN_NAMES = ("string", "_controller_hook")
+    FUNCTION = "get_value"
+    CATEGORY = category + subcategories["io"]
+
+    def get_value(self, input_name, value):
+        return (value, None)
+
+
+class NilorUserInput_Int:
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "input_name": (
+                    "STRING",
+                    {"default": "my_int_input", "multiline": False},
+                ),
+                "value": ("INT", {"default": 0}),
+            }
+        }
+
+    RETURN_TYPES = ("INT", CONTROLLER_HOOK)
+    RETURN_NAMES = ("int", "_controller_hook")
+    FUNCTION = "get_value"
+    CATEGORY = category + subcategories["io"]
+
+    def get_value(self, input_name, value):
+        return (value, None)
+
+
+class NilorUserInput_Float:
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "input_name": (
+                    "STRING",
+                    {"default": "my_float_input", "multiline": False},
+                ),
+                "value": ("FLOAT", {"default": 0.0}),
+            }
+        }
+
+    RETURN_TYPES = ("FLOAT", CONTROLLER_HOOK)
+    RETURN_NAMES = ("float", "_controller_hook")
+    FUNCTION = "get_value"
+    CATEGORY = category + subcategories["io"]
+
+    def get_value(self, input_name, value):
+        return (value, None)
+
+
+class NilorUserInput_Boolean:
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "input_name": (
+                    "STRING",
+                    {"default": "my_bool_input", "multiline": False},
+                ),
+                "value": ("BOOLEAN", {"default": False}),
+            }
+        }
+
+    RETURN_TYPES = ("BOOLEAN", CONTROLLER_HOOK)
+    RETURN_NAMES = ("boolean", "_controller_hook")
+    FUNCTION = "get_value"
+    CATEGORY = category + subcategories["io"]
+
+    def get_value(self, input_name, value):
+        return (value, None)
+
+
+NODE_CLASS_MAPPINGS = {
+    "NilorUserInput_String": NilorUserInput_String,
+    "NilorUserInput_Int": NilorUserInput_Int,
+    "NilorUserInput_Float": NilorUserInput_Float,
+    "NilorUserInput_Boolean": NilorUserInput_Boolean,
+}
+
+NODE_DISPLAY_NAME_MAPPINGS = {
+    "NilorUserInput_String": "👺 User Input (String)",
+    "NilorUserInput_Int": "👺 User Input (Int)",
+    "NilorUserInput_Float": "👺 User Input (Float)",
+    "NilorUserInput_Boolean": "👺 User Input (Boolean)",
+}

utils.py

@@ -15,8 +15,9 @@ def numpy2pil(image: np.ndarray, mode=None):
 
 
 ## Helper function equivalent to Mikey's pil2tensor
-    #def pil2tensor(self, image):
-    #    return torch.from_numpy(np.array(image).astype(np.float32) / 255.0).unsqueeze(0)
+# def pil2tensor(self, image):
+#    return torch.from_numpy(np.array(image).astype(np.float32) / 255.0).unsqueeze(0)
+
 
 def pil2tensor(image: Image.Image):
     return torch.from_numpy(pil2numpy(image)).unsqueeze(0)

web/js/controllers.js

@@ -0,0 +1,272 @@
+import { app } from "/scripts/app.js";
+
+// NilorPreset dynamic inputs extension
+// Adds a new empty _input_hook_N slot whenever the last slot gets connected, up to a hard cap
+
+const MAX_INPUTS = 32;
+// Preset controller constants
+const CLASS_TYPE = "NilorPreset";
+const INPUT_PREFIX = "_preset_hook_";
+
+function isTargetNode(node) {
+    return node && (node.comfyClass === CLASS_TYPE || node.type === CLASS_TYPE);
+}
+
+function countHookInputs(node) {
+    return (node.inputs || []).filter((i) => i && i.name?.startsWith(INPUT_PREFIX)).length;
+}
+
+function nextInputName(node) {
+    let index = 1;
+    while (index <= MAX_INPUTS) {
+        const key = `${INPUT_PREFIX}${index}`;
+        if (!node.inputs || !node.inputs.find((i) => i.name === key)) {
+            return key;
+        }
+        index++;
+    }
+    return null;
+}
+
+function resizeNode(node) {
+    try {
+        const size = node.computeSize();
+        node.onResize?.(size);
+        app.graph?.setDirtyCanvas(true, true);
+    } catch (_) {}
+}
+
+function ensureAtLeastOneSlot(node) {
+    if (!isTargetNode(node)) return;
+    if (countHookInputs(node) === 0) {
+        const name = `${INPUT_PREFIX}1`;
+        node.addInput(name, "CONTROLLER_HOOK");
+        resizeNode(node);
+    }
+}
+
+function growIfLastLinked(node) {
+    if (!isTargetNode(node)) return;
+    const inputs = (node.inputs || []).filter((i) => i && i.name?.startsWith(INPUT_PREFIX));
+    if (inputs.length === 0) return;
+    const last = inputs[inputs.length - 1];
+    const lastIsLinked = !!last.link;
+    if (lastIsLinked && inputs.length < MAX_INPUTS) {
+        const name = nextInputName(node);
+        if (name) {
+            node.addInput(name, "CONTROLLER_HOOK");
+            resizeNode(node);
+        }
+    }
+}
+
+function shrinkTrailingUnlinked(node) {
+    if (!isTargetNode(node)) return;
+    const allInputs = node.inputs || [];
+    // Collect indices of hook inputs
+    const hookIndices = [];
+    for (let i = 0; i < allInputs.length; i++) {
+        const inp = allInputs[i];
+        if (inp && inp.name && inp.name.startsWith(INPUT_PREFIX)) {
+            hookIndices.push(i);
+        }
+    }
+    if (hookIndices.length <= 1) return; // always keep at least one
+
+    // Find last linked among hook inputs (by position in hookIndices)
+    let lastLinkedPos = -1;
+    for (let pos = 0; pos < hookIndices.length; pos++) {
+        const idx = hookIndices[pos];
+        if (allInputs[idx]?.link) lastLinkedPos = pos;
+    }
+
+    const targetHookCount = lastLinkedPos >= 0 ? lastLinkedPos + 1 : 1;
+
+    // Remove trailing unlinked beyond targetHookCount
+    for (let pos = hookIndices.length - 1; pos >= targetHookCount; pos--) {
+        const idx = hookIndices[pos];
+        const input = node.inputs[idx];
+        if (input && !input.link) {
+            try {
+                node.removeInput(idx);
+            } catch (e) {
+                console.warn("nilor-preset-dynamic-inputs removeInput error", e);
+                break;
+            }
+        } else {
+            break;
+        }
+    }
+    resizeNode(node);
+}
+
+app.registerExtension({
+    name: "comfy.nilor-nodes.userinputPreset",
+
+    // Ensure compatibility with saved/loaded graphs
+    afterConfigureGraph(graph) {
+        try {
+            (graph?._nodes || graph?.nodes || []).forEach((n) => {
+                if (isTargetNode(n)) {
+                    ensureAtLeastOneSlot(n);
+                    shrinkTrailingUnlinked(n);
+                    growIfLastLinked(n);
+                }
+            });
+        } catch (e) {
+            console.warn("nilor-preset-dynamic-inputs afterConfigureGraph error", e);
+        }
+    },
+
+    // Patch the prototype so we always react to connection changes
+    async beforeRegisterNodeDef(nodeType, nodeData, appInstance) {
+        if (nodeData?.name !== CLASS_TYPE) return;
+        const original = nodeType.prototype.onConnectionsChange;
+        nodeType.prototype.onConnectionsChange = function (type, index, connected, link_info) {
+            if (typeof original === "function") {
+                original.apply(this, arguments);
+            }
+            try {
+                shrinkTrailingUnlinked(this);
+                growIfLastLinked(this);
+            } catch (e) {
+                console.warn("nilor-preset-dynamic-inputs onConnectionsChange error", e);
+            }
+        };
+    },
+
+    nodeCreated(node) {
+        if (!isTargetNode(node)) return;
+        ensureAtLeastOneSlot(node);
+        shrinkTrailingUnlinked(node);
+        growIfLastLinked(node);
+    },
+});
+
+
+// NilorGroup dynamic inputs extension (mirrors preset behavior)
+const GROUP_CLASS_TYPE = "NilorGroup";
+const GROUP_INPUT_PREFIX = "_group_hook_";
+
+function isGroupNode(node) {
+    return node && (node.comfyClass === GROUP_CLASS_TYPE || node.type === GROUP_CLASS_TYPE);
+}
+
+function countGroupHookInputs(node) {
+    return (node.inputs || []).filter((i) => i && i.name?.startsWith(GROUP_INPUT_PREFIX)).length;
+}
+
+function nextGroupInputName(node) {
+    let index = 1;
+    while (index <= MAX_INPUTS) {
+        const key = `${GROUP_INPUT_PREFIX}${index}`;
+        if (!node.inputs || !node.inputs.find((i) => i.name === key)) {
+            return key;
+        }
+        index++;
+    }
+    return null;
+}
+
+function ensureAtLeastOneGroupSlot(node) {
+    if (!isGroupNode(node)) return;
+    if (countGroupHookInputs(node) === 0) {
+        const name = `${GROUP_INPUT_PREFIX}1`;
+        node.addInput(name, "CONTROLLER_HOOK");
+        resizeNode(node);
+    }
+}
+
+function growGroupIfLastLinked(node) {
+    if (!isGroupNode(node)) return;
+    const inputs = (node.inputs || []).filter((i) => i && i.name?.startsWith(GROUP_INPUT_PREFIX));
+    if (inputs.length === 0) return;
+    const last = inputs[inputs.length - 1];
+    const lastIsLinked = !!last.link;
+    if (lastIsLinked && inputs.length < MAX_INPUTS) {
+        const name = nextGroupInputName(node);
+        if (name) {
+            node.addInput(name, "CONTROLLER_HOOK");
+            resizeNode(node);
+        }
+    }
+}
+
+function shrinkGroupTrailingUnlinked(node) {
+    if (!isGroupNode(node)) return;
+    const allInputs = node.inputs || [];
+    const hookIndices = [];
+    for (let i = 0; i < allInputs.length; i++) {
+        const inp = allInputs[i];
+        if (inp && inp.name && inp.name.startsWith(GROUP_INPUT_PREFIX)) {
+            hookIndices.push(i);
+        }
+    }
+    if (hookIndices.length <= 1) return;
+
+    let lastLinkedPos = -1;
+    for (let pos = 0; pos < hookIndices.length; pos++) {
+        const idx = hookIndices[pos];
+        if (allInputs[idx]?.link) lastLinkedPos = pos;
+    }
+
+    const targetHookCount = lastLinkedPos >= 0 ? lastLinkedPos + 1 : 1;
+
+    for (let pos = hookIndices.length - 1; pos >= targetHookCount; pos--) {
+        const idx = hookIndices[pos];
+        const input = node.inputs[idx];
+        if (input && !input.link) {
+            try {
+                node.removeInput(idx);
+            } catch (e) {
+                console.warn("nilor-group-dynamic-inputs removeInput error", e);
+                break;
+            }
+        } else {
+            break;
+        }
+    }
+    resizeNode(node);
+}
+
+app.registerExtension({
+    name: "comfy.nilor-nodes.userinputGroup",
+
+    afterConfigureGraph(graph) {
+        try {
+            (graph?._nodes || graph?.nodes || []).forEach((n) => {
+                if (isGroupNode(n)) {
+                    ensureAtLeastOneGroupSlot(n);
+                    shrinkGroupTrailingUnlinked(n);
+                    growGroupIfLastLinked(n);
+                }
+            });
+        } catch (e) {
+            console.warn("nilor-group-dynamic-inputs afterConfigureGraph error", e);
+        }
+    },
+
+    async beforeRegisterNodeDef(nodeType, nodeData, appInstance) {
+        if (nodeData?.name !== GROUP_CLASS_TYPE) return;
+        const original = nodeType.prototype.onConnectionsChange;
+        nodeType.prototype.onConnectionsChange = function (type, index, connected, link_info) {
+            if (typeof original === "function") {
+                original.apply(this, arguments);
+            }
+            try {
+                shrinkGroupTrailingUnlinked(this);
+                growGroupIfLastLinked(this);
+            } catch (e) {
+                console.warn("nilor-group-dynamic-inputs onConnectionsChange error", e);
+            }
+        };
+    },
+
+    nodeCreated(node) {
+        if (!isGroupNode(node)) return;
+        ensureAtLeastOneGroupSlot(node);
+        shrinkGroupTrailingUnlinked(node);
+        growGroupIfLastLinked(node);
+    },
+});
+

web/js/media_stream.js

@@ -0,0 +1,63 @@
+import { app } from "/scripts/app.js";
+
+function toggleFramerateWidget(node, show) {
+    const framerateWidget = node.widgets.find((w) => w.name === "framerate");
+    if (framerateWidget) {
+        framerateWidget.hidden = !show;
+        // This is a hack to force the node to redraw.
+        //const size = node.computeSize();
+        //node.onResize?.(size);
+    }
+}
+
+function hideWidgets(node, widgetNames) {
+    widgetNames.forEach(name => {
+        const widget = node.widgets.find((w) => w.name === name);
+        if (widget) {
+            widget.hidden = true;
+        }
+    });
+}
+
+app.registerExtension({
+    name: "comfy.nilor-nodes.mediaStream",
+    nodeCreated(node) {
+        if (node.comfyClass === "MediaStreamOutput") {
+            // Hide system inputs by default
+            hideWidgets(node, [
+                "content_id",
+                "venue",
+                "canvas",
+                "scene",
+                "presigned_upload_url",
+                "job_completions_queue_url",
+                "output_object_keys"
+            ]);
+
+            const formatWidget = node.widgets.find((w) => w.name === "format");
+
+            // Initial toggle for framerate based on the default format value
+            toggleFramerateWidget(node, formatWidget.value === "mp4");
+
+            // Store original callback to chain it
+            const originalCallback = formatWidget.callback;
+
+            formatWidget.callback = function (value) {
+                toggleFramerateWidget(node, value === "mp4");
+
+                // Recalculate node size after toggling widgets
+                const size = node.computeSize();
+                node.onResize?.(size);
+                
+                if (originalCallback) {
+                    return originalCallback.apply(this, arguments);
+                }
+            };
+        }
+
+        if (node.comfyClass === "MediaStreamInput") {
+            // Hide system inputs by default
+            hideWidgets(node, ["presigned_download_url"]);
+        }
+    },
+});

worker_consumer.py

@@ -0,0 +1,441 @@
+"""
+Worker Consumer Service for ComfyUI
+
+This script runs as a continuous background service on each ComfyUI worker.
+Its purpose is to poll the `jobs_to_process` SQS queue for new jobs,
+submit them to the local ComfyUI server, and manage the message lifecycle.
+It also listens to the ComfyUI websocket to send a "running" status update
+at the precise moment that job execution begins.
+"""
+
+import os
+import json
+import logging
+import asyncio
+import aiohttp
+import websockets
+from aiobotocore.session import get_session
+from dotenv import load_dotenv
+from botocore.exceptions import EndpointConnectionError
+
+# --- Load Environment Variables ---
+# Load from the .env file in the same directory
+current_dir = os.path.dirname(os.path.abspath(__file__))
+dotenv_path = os.path.join(current_dir, ".env")
+if os.path.exists(dotenv_path):
+    load_dotenv(dotenv_path=dotenv_path)
+    logging.info(
+        f"✅\u2009 Nilor-Nodes: Loaded environment variables from {dotenv_path}"
+    )
+else:
+    logging.info(
+        "⚠️\u2009 Nilor-Nodes: No .env file found, relying on shell environment variables."
+    )
+
+# --- Configuration ---
+SQS_ENDPOINT_URL = os.getenv("SQS_ENDPOINT_URL", "http://localhost:9324")
+SQS_JOBS_TO_PROCESS_QUEUE_NAME = os.getenv(
+    "SQS_JOBS_TO_PROCESS_QUEUE_NAME", "jobs_to_process"
+)
+SQS_JOB_STATUS_UPDATES_QUEUE_NAME = os.getenv(
+    "SQS_JOB_STATUS_UPDATES_QUEUE_NAME", "job_status_updates"
+)
+COMFYUI_API_URL = os.getenv("COMFYUI_API_URL", "http://127.0.0.1:8188") + "/prompt"
+COMFYUI_WS_URL = os.getenv("COMFYUI_WS_URL", "ws://127.0.0.1:8188") + "/ws"
+AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID", "local")
+AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY", "local")
+AWS_DEFAULT_REGION = os.getenv("AWS_DEFAULT_REGION", "us-east-1")
+POLL_WAIT_TIME_SECONDS = 20  # SQS Long Polling
+MAX_MESSAGES = 1
+
+# --- Setup Logging ---
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+)
+
+
+class WorkerConsumer:
+    def __init__(self):
+        self.session = get_session()
+        self.prompt_id_to_content_id_map = {}
+        self.sent_running_status_prompts = set()
+        self.content_context_by_content_id = {}
+        self.jobs_queue_url = None
+        self.status_updates_queue_url = None
+        self.http_session = None
+
+    async def _initialize_sqs(self):
+        """Initializes SQS queue URLs. Returns True on success, False on failure."""
+        async with self.session.create_client(
+            "sqs",
+            region_name=AWS_DEFAULT_REGION,
+            endpoint_url=SQS_ENDPOINT_URL,
+            aws_access_key_id=AWS_ACCESS_KEY_ID,
+            aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
+        ) as client:
+            try:
+                self.jobs_queue_url = await self._get_queue_url(
+                    client, SQS_JOBS_TO_PROCESS_QUEUE_NAME
+                )
+                self.status_updates_queue_url = await self._get_queue_url(
+                    client, SQS_JOB_STATUS_UPDATES_QUEUE_NAME
+                )
+                return True
+            except EndpointConnectionError as e:
+                # Quiet the noisy traceback by logging a concise warning instead
+                logging.warning(
+                    f"⚠️\u2009 Nilor-Nodes (worker_consumer): SQS endpoint is unreachable at {SQS_ENDPOINT_URL}: {e}. "
+                )
+                return False
+            except Exception as e:
+                logging.error(
+                    f"⚠️\u2009 Nilor-Nodes (worker_consumer): Failed to initialize SQS queues: {e}"
+                )
+                return False
+
+    async def _get_queue_url(self, client, queue_name):
+        """Retrieves the SQS queue URL."""
+        try:
+            response = await client.get_queue_url(QueueName=queue_name)
+            return response["QueueUrl"]
+        except client.exceptions.QueueDoesNotExist:
+            logging.error(
+                f"⚠️\u2009 Nilor-Nodes (worker_consumer): SQS queue '{queue_name}' does not exist."
+            )
+            raise
+
+    async def listen_for_comfy_events(self):
+        while True:
+            try:
+                async with websockets.connect(COMFYUI_WS_URL) as websocket:
+                    logging.info(
+                        f"ℹ️\u2009 Nilor-Nodes (worker_consumer): Connected to ComfyUI websocket at {COMFYUI_WS_URL}"
+                    )
+                    while True:
+                        message = await websocket.recv()
+                        if isinstance(message, str):
+                            try:
+                                event = json.loads(message)
+                                event_type = event.get("type")
+                                data = event.get("data", {})
+                                prompt_id = data.get("prompt_id")
+
+                                if not prompt_id and "sid" in data:
+                                    prompt_id = data["sid"]
+
+                                if not prompt_id:
+                                    continue
+
+                                # Use the first progress event as a signal that the job is running.
+                                if (
+                                    event_type in ["progress", "progress_state"]
+                                    and prompt_id in self.prompt_id_to_content_id_map
+                                    and prompt_id
+                                    not in self.sent_running_status_prompts
+                                ):
+                                    content_id = self.prompt_id_to_content_id_map[
+                                        prompt_id
+                                    ]
+                                    ctx = self.content_context_by_content_id.get(
+                                        content_id, {}
+                                    )
+                                    policy = ctx.get("status_policy") or {}
+                                    running_status = policy.get(
+                                        "running_status", "running"
+                                    )
+                                    logging.info(
+                                        f"ℹ️\u2009 Nilor-Nodes (worker_consumer): Execution started for prompt_id {prompt_id} (content_id: {content_id}) via '{event_type}' event. Sending '{running_status}' status."
+                                    )
+                                    await self._send_status_update(
+                                        content_id,
+                                        running_status,
+                                        ctx.get("venue"),
+                                        ctx.get("canvas"),
+                                        ctx.get("scene"),
+                                    )
+                                    self.sent_running_status_prompts.add(prompt_id)
+
+                                # Handle execution errors
+                                elif event_type == "execution_error":
+                                    logging.error(
+                                        f"🛑\u2009 Nilor-Nodes (worker_consumer): Received execution error for prompt_id {prompt_id}: {data}"
+                                    )
+                                    if prompt_id in self.prompt_id_to_content_id_map:
+                                        content_id = (
+                                            self.prompt_id_to_content_id_map.pop(
+                                                prompt_id
+                                            )
+                                        )
+                                        ctx = self.content_context_by_content_id.get(
+                                            content_id, {}
+                                        )
+                                        policy = ctx.get("status_policy") or {}
+                                        fail_status = policy.get(
+                                            "fail_status", "failed"
+                                        )
+                                        try:
+                                            await self._send_status_update(
+                                                content_id,
+                                                fail_status,
+                                                ctx.get("venue"),
+                                                ctx.get("canvas"),
+                                                ctx.get("scene"),
+                                            )
+                                        except Exception:
+                                            pass
+                                        self.content_context_by_content_id.pop(
+                                            content_id, None
+                                        )
+                                    self.sent_running_status_prompts.discard(prompt_id)
+
+                                # Log successful execution
+                                elif event_type == "executed":
+                                    logging.info(
+                                        f"✅ Nilor-Nodes (worker_consumer): Prompt {prompt_id} executed successfully according to websocket event. Final node is responsible for sending completion message."
+                                    )
+                                    if prompt_id in self.prompt_id_to_content_id_map:
+                                        content_id = (
+                                            self.prompt_id_to_content_id_map.pop(
+                                                prompt_id
+                                            )
+                                        )
+                                        self.content_context_by_content_id.pop(
+                                            content_id, None
+                                        )
+                                    self.sent_running_status_prompts.discard(prompt_id)
+
+                                elif event_type not in ["progress", "progress_state"]:
+                                    logging.info(
+                                        f"ℹ️\u2009 Nilor-Nodes (worker_consumer): Received ComfyUI websocket event of type '{event_type}': {data}"
+                                    )
+
+                            except json.JSONDecodeError:
+                                logging.debug(
+                                    "⚠️\u2009 Nilor-Nodes (worker_consumer): Received non-JSON text message from websocket, ignoring."
+                                )
+                        else:
+                            logging.debug(
+                                "⚠️\u2009 Nilor-Nodes (worker_consumer): Received binary message from websocket, ignoring."
+                            )
+            except (
+                websockets.exceptions.ConnectionClosedError,
+                ConnectionRefusedError,
+            ) as e:
+                logging.warning(
+                    f"🛑\u2009 Nilor-Nodes (worker_consumer): ComfyUI websocket connection failed: {e}. Retrying in 5 seconds..."
+                )
+                await asyncio.sleep(5)
+            except Exception as e:
+                logging.error(
+                    f"🛑\u2009 Nilor-Nodes (worker_consumer): An unexpected error occurred in the websocket listener: {e}",
+                    exc_info=True,
+                )
+                await asyncio.sleep(10)
+
+    async def consume_loop(self):
+        """The main loop to continuously poll for and process messages.
+        Keeps retrying SQS initialization and polling if the endpoint is down.
+        """
+
+        # Start the websocket listener in the background immediately
+        listener_task = asyncio.create_task(self.listen_for_comfy_events())
+
+        try:
+            while True:
+                # Ensure SQS is initialized; if not, keep attempting to initialize
+                if self.jobs_queue_url is None or self.status_updates_queue_url is None:
+                    initialized = await self._initialize_sqs()
+                    if not initialized:
+                        logging.warning(
+                            "⚠️\u2009 Nilor-Nodes (worker_consumer): SQS initialization failed. Retrying in 10 seconds..."
+                        )
+                        await asyncio.sleep(10)
+                        continue
+                    logging.info(
+                        f"ℹ️\u2009 Nilor-Nodes (worker_consumer): Starting worker consumer. Polling queue: {self.jobs_queue_url}"
+                    )
+
+                logging.debug(
+                    "ℹ️\u2009 Nilor-Nodes (worker_consumer): Polling for messages..."
+                )
+                try:
+                    async with self.session.create_client(
+                        "sqs",
+                        region_name=AWS_DEFAULT_REGION,
+                        endpoint_url=SQS_ENDPOINT_URL,
+                        aws_access_key_id=AWS_ACCESS_KEY_ID,
+                        aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
+                    ) as client:
+                        response = await client.receive_message(
+                            QueueUrl=self.jobs_queue_url,
+                            MaxNumberOfMessages=MAX_MESSAGES,
+                            WaitTimeSeconds=POLL_WAIT_TIME_SECONDS,
+                        )
+
+                    messages = response.get("Messages", [])
+                    if not messages:
+                        logging.debug(
+                            "ℹ️\u2009 Nilor-Nodes (worker_consumer): No messages received."
+                        )
+                        continue
+
+                    for message in messages:
+                        try:
+                            await self.process_message(message)
+
+                            # On successful processing, delete the message
+                            async with self.session.create_client(
+                                "sqs",
+                                region_name=AWS_DEFAULT_REGION,
+                                endpoint_url=SQS_ENDPOINT_URL,
+                                aws_access_key_id=AWS_ACCESS_KEY_ID,
+                                aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
+                            ) as client:
+                                await client.delete_message(
+                                    QueueUrl=self.jobs_queue_url,
+                                    ReceiptHandle=message["ReceiptHandle"],
+                                )
+                            logging.info(
+                                f"ℹ️\u2009 Nilor-Nodes (worker_consumer): Deleted message {message['MessageId']} from queue."
+                            )
+                        except json.JSONDecodeError:
+                            # This is a poison pill message, log it but don't retry.
+                            # It will be moved to the DLQ after enough failed receives.
+                            logging.error(
+                                f"🛑\u2009 Nilor-Nodes (worker_consumer): Message {message['MessageId']} is a poison pill (JSON decode failed) and will be ignored."
+                            )
+                        except Exception as e:
+                            logging.error(
+                                f"🛑\u2009 Nilor-Nodes (worker_consumer): Processing failed for message {message['MessageId']}: {e}. It will be returned to the queue for retry."
+                            )
+
+                except EndpointConnectionError as e:
+                    # Lost connection to SQS; reset and re-initialize on next loop
+                    logging.warning(
+                        f"⚠️\u2009 Nilor-Nodes (worker_consumer): Lost connection to SQS at {SQS_ENDPOINT_URL}: {e}. Will retry initialization in 10 seconds."
+                    )
+                    self.jobs_queue_url = None
+                    self.status_updates_queue_url = None
+                    await asyncio.sleep(10)
+                except Exception as e:
+                    logging.error(
+                        f"🛑\u2009 Nilor-Nodes (worker_consumer): An error occurred in the consume loop: {e}"
+                    )
+                    await asyncio.sleep(10)  # Wait before retrying
+        finally:
+            listener_task.cancel()
+            await asyncio.gather(listener_task, return_exceptions=True)
+            logging.info(
+                "⚠️\u2009 Nilor-Nodes (worker_consumer): Websocket listener stopped."
+            )
+
+    async def process_message(self, message):
+        """Processes a single SQS message."""
+        logging.info(
+            f"ℹ️\u2009 Nilor-Nodes (worker_consumer): Processing message: {message['MessageId']}"
+        )
+
+        try:
+            body = json.loads(message["Body"])
+            # SQS messages are often double-encoded, with the actual payload inside a 'Message' key.
+            if "Message" in body:
+                job_payload = json.loads(body["Message"])
+            else:
+                job_payload = body
+
+            content_id = job_payload.get("content_id")
+
+            # Validate that the payload has the required keys before submitting.
+            if not content_id or "prompt" not in job_payload:
+                logging.error(
+                    f"🛑\u2009 Nilor-Nodes (worker_consumer): Invalid message format: missing 'content_id' or 'prompt'. Payload: {job_payload}"
+                )
+                return
+
+            # Submit to ComfyUI
+            await self._submit_job_to_comfyui(content_id, job_payload)
+
+            # Cache context for subsequent status updates
+            try:
+                self.content_context_by_content_id[content_id] = {
+                    "venue": job_payload.get("venue"),
+                    "canvas": job_payload.get("canvas"),
+                    "scene": job_payload.get("scene"),
+                    "status_policy": job_payload.get("status_policy") or {},
+                }
+            except Exception:
+                self.content_context_by_content_id[content_id] = {}
+
+        except Exception as e:
+            logging.error(
+                f"🛑\u2009 Nilor-Nodes (worker_consumer): An unexpected error occurred while processing message: {e}. It will be retried."
+            )
+            # Re-raise to prevent deletion from queue if we want SQS to handle retry
+            raise
+
+    async def _submit_job_to_comfyui(self, content_id, workflow_data):
+        """Submits a single job to the ComfyUI API."""
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    COMFYUI_API_URL, json=workflow_data, timeout=30
+                ) as response:
+                    response.raise_for_status()
+                    response_json = await response.json()
+                    prompt_id = response_json.get("prompt_id")
+                    logging.info(
+                        f"✅ Nilor-Nodes (worker_consumer): Successfully submitted job to ComfyUI. Prompt ID: {prompt_id}"
+                    )
+                    self.prompt_id_to_content_id_map[prompt_id] = content_id
+
+            # No need to delete here, the consume_loop handles message deletion
+        except aiohttp.ClientError as e:
+            logging.error(
+                f"🛑\u2009 Nilor-Nodes (worker_consumer): Failed to submit job to ComfyUI: {e}. Message will be retried."
+            )
+        except (json.JSONDecodeError, KeyError) as e:
+            logging.error(
+                f"🛑\u2009 Nilor-Nodes (worker_consumer): Failed to parse ComfyUI response: {e}. Discarding malformed response."
+            )
+        except Exception as e:
+            logging.error(
+                f"🛑\u2009 Nilor-Nodes (worker_consumer): An unexpected error occurred while submitting job to ComfyUI: {e}",
+                exc_info=True,
+            )
+
+    async def _send_status_update(
+        self, content_id, status, venue=None, canvas=None, scene=None
+    ):
+        try:
+            body = {"content_id": content_id, "status": status}
+            if venue is not None:
+                body["venue"] = venue
+            if canvas is not None:
+                body["canvas"] = canvas
+            if scene is not None:
+                body["scene"] = scene
+            message_body = json.dumps(body)
+            async with self.session.create_client(
+                "sqs",
+                region_name=AWS_DEFAULT_REGION,
+                endpoint_url=SQS_ENDPOINT_URL,
+                aws_access_key_id=AWS_ACCESS_KEY_ID,
+                aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
+            ) as client:
+                await client.send_message(
+                    QueueUrl=self.status_updates_queue_url, MessageBody=message_body
+                )
+            logging.info(
+                f"✅ Nilor-Nodes (worker_consumer): Sent status update for content {content_id}: {status}"
+            )
+        except Exception as e:
+            logging.error(
+                f"🛑\u2009 Nilor-Nodes (worker_consumer): Failed to send status update for content {content_id}: {e}",
+                exc_info=True,
+            )
+
+
+async def consume_jobs():
+    """Entry point function to be called in a background thread."""
+    consumer = WorkerConsumer()
+    await consumer.consume_loop()