chore: remove smoke-test artifacts, gitignore output/

The URL→request_id lookup only searched cached results, so outputs
fetched via Submit→Collect or request-id recovery had no provenance
(sidecars were written with null endpoint/request). Add an explicit
request_urls table populated on every successful fetch path; the lookup
checks it first and falls back to the result-JSON scan.

Verified against the live fal API: full 11-check smoke suite passes,
including the Save→Provenance-from-File round trip on a real CDN file.
Also confirmed live: /account/billing returns 403 for non-admin keys
(handled gracefully with a hint) and the cache/dedup/free-recovery
ledger semantics hold against real requests.

.gitignore

@@ -174,3 +174,4 @@ memory-bank/
 .DS_Store
 .claude/
 Node-Docs/
+output/

README.md

@@ -17,6 +17,7 @@ Custom nodes that bring the entire [fal.ai](https://fal.ai) catalog into ComfyUI
   - [Vision Language Models (VLMs)](#vision-language-models-vlms)
 - [Auto-Generated Nodes](#auto-generated-nodes)
 - [Platform Utilities](#platform-utilities)
+- [Utility Nodes](#utility-nodes-24)
 - [Generated Model List](#generated-model-list)
 - [Registry Maintenance](#registry-maintenance)
 - [Troubleshooting](#troubleshooting)
@@ -278,6 +279,16 @@ All frontend features degrade silently on older ComfyUI versions.
 
 fal result URLs eventually expire from the CDN. This node downloads any result URL into your ComfyUI `output/` directory (subfolders + collision-safe numbering) so generations are persisted with your other outputs.
 
+## Utility Nodes (2.4)
+
+Twenty nodes under `FAL/Utils` covering everything between your assets and a fal endpoint — no other packs needed:
+
+- **Dataset** (`FAL/Utils/Dataset`): *Images → Training ZIP URL* (standard LoRA caption layout), *Folder → ZIP URL*, *Video → Frame Dataset ZIP URL*, *Batch Caption Images* (parallel VLM captioning). The full pipeline: Load Image Folder → Batch Caption → Images→ZIP → any trainer node.
+- **Load** (`FAL/Utils/Load`): *Load Image from URL* (multi-URL batching), *Load Audio from URL*, *Load Image Folder*, *Upload Folder as ZIP URL*.
+- **Video** (`FAL/Utils/Video`): *Extract Frames* (efficient last-frame seek — chain into image-to-video for endless extension), *Trim* (keyframe remux, no re-encode), *Concat* (auto resolution/fps normalize), *Mux Audio + Video* (marry TTS/music to generated video), *Video → Audio*.
+- **Image** (`FAL/Utils/Image`): *Image Grid with Labels*, *Resize to fal Preset* (cover/contain/stretch to the standard image_size dims), *Image ↔ Base64*.
+- **Data** (`FAL/Utils/Data`): *JSON Extract* (dot/bracket path queries against any `result_json` output), *Prompt Lines* (cycling line picker), *Text Template*.
+
 ## Generated Model List
 
 The full list of auto-generated model nodes, grouped by category (largest first). Click a category to expand it. This section is regenerated with `python scripts/build_readme.py`.

__init__.py

@@ -11,6 +11,11 @@ node_list = [
     "platform_node",
     "billing_node",
     "inbox_node",
+    "util_dataset_node",
+    "util_media_in_node",
+    "util_video_node",
+    "util_image_node",
+    "util_data_node",
 ]
 
 NODE_CLASS_MAPPINGS = {}

nodes/fal_utils.py

@@ -9,6 +9,7 @@ The implementations now live in the ``nodes/utils`` package.
 
 from .utils import (
     ApiHandler,
+    ArchiveUtils,
     BillingUtils,
     FalApiError,
     FalConfig,
@@ -25,6 +26,7 @@ from .utils import (
 
 __all__ = [
     "ApiHandler",
+    "ArchiveUtils",
     "BillingUtils",
     "FalApiError",
     "FalConfig",

nodes/trainer_node.py

@@ -1,46 +1,15 @@
-import os
-import tempfile
-import zipfile
-
-import torch
-from PIL import Image
-
-from .fal_utils import ApiHandler, FalConfig, ImageUtils
+from .fal_utils import ApiHandler, ArchiveUtils, FalConfig
 
 # Initialize FalConfig
 fal_config = FalConfig()
 
 
 def create_zip_from_images(images):
-    """Create a zip file from a list of images."""
+    """Create a zip file from a list of images and upload it (returns the URL)."""
     try:
-        with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as temp_zip:
-            with zipfile.ZipFile(temp_zip, "w") as zf:
-                for idx, img_tensor in enumerate(images):
-                    # Convert tensor to PIL Image
-                    if isinstance(img_tensor, torch.Tensor):
-                        # Convert to numpy and scale to 0-255 range
-                        img_np = (img_tensor.cpu().numpy() * 255).astype("uint8")
-                        # Handle different tensor formats
-                        if img_np.shape[0] == 3:  # If in format (C, H, W)
-                            img_np = img_np.transpose(1, 2, 0)
-                        img = Image.fromarray(img_np)
-                    else:
-                        img = img_tensor
-
-                    # Save image to temporary file
-                    with tempfile.NamedTemporaryFile(
-                        suffix=".png", delete=False
-                    ) as temp_img:
-                        img.save(temp_img, format="PNG")
-                        temp_img_path = temp_img.name
-
-                    # Add to zip file
-                    zf.write(temp_img_path, f"image_{idx}.png")
-                    os.unlink(temp_img_path)
-
-            # Upload the zip through the shared utility (raises on failure)
-            return ImageUtils.upload_file(temp_zip.name)
+        zip_path = ArchiveUtils.zip_images(images)
+        # Upload the zip through the shared utility (raises on failure)
+        return ArchiveUtils.upload_zip(zip_path)
     except Exception as e:
         return ApiHandler.handle_text_generation_error(
             "flux-lora-fast-training", f"Failed to create zip file: {str(e)}"

nodes/util_data_node.py

@@ -0,0 +1,259 @@
+"""Data utility nodes: JSON path extraction, prompt line cycling, text templating."""
+
+from __future__ import annotations
+
+import json
+import re
+from typing import Any
+
+from .fal_utils import FalApiError, logger
+
+_CATEGORY = "FAL/Utils/Data"
+
+_MAX_INDEX = 2**31 - 1
+_MISSING = object()
+
+# a path segment is an optional key name followed by zero or more [N] indices
+_PATH_SEGMENT = re.compile(r"^([^\[\]]*)((?:\[\d+\])*)$")
+_BRACKET_INDEX = re.compile(r"\[(\d+)\]")
+
+
+def _tokenize_path(path: str) -> list[Any]:
+    """Split a dot/bracket path into str keys and int indices. No eval."""
+    tokens: list[Any] = []
+    for part in path.split("."):
+        segment = part.strip()
+        if not segment:
+            continue
+        match = _PATH_SEGMENT.match(segment)
+        if match is None:
+            # contract: anything that can't resolve returns the default,
+            # a malformed segment included — it can never match a key anyway
+            logger.debug("FalJSONExtract: unparseable path segment %r in %r", segment, path)
+            return None
+        name, brackets = match.group(1), match.group(2)
+        if name:
+            tokens.append(name)
+        tokens.extend(int(index) for index in _BRACKET_INDEX.findall(brackets))
+    return tokens
+
+
+def _value_to_bool(value: Any) -> bool:
+    """Truthiness with JSON-string awareness: "false"/"0"/"no"/"" are False."""
+    if isinstance(value, str):
+        return value.strip().lower() not in ("", "false", "0", "no", "none", "null")
+    return bool(value)
+
+
+def _walk_path(value: Any, tokens: list[Any]) -> Any:
+    """Follow tokens through nested dicts/lists; return _MISSING when absent."""
+    current = value
+    for token in tokens:
+        index = token if isinstance(token, int) else None
+        if index is None and isinstance(current, list) and str(token).isdigit():
+            index = int(token)  # bare integer segment indexing an array
+        if index is not None:
+            if isinstance(current, list) and 0 <= index < len(current):
+                current = current[index]
+            else:
+                return _MISSING
+        elif isinstance(current, dict) and token in current:
+            current = current[token]
+        else:
+            return _MISSING
+    return current
+
+
+def _value_to_text(value: Any) -> str:
+    """Strings pass through; everything else is re-serialized as JSON."""
+    if isinstance(value, str):
+        return value
+    return json.dumps(value)
+
+
+def _value_to_number(value: Any) -> float:
+    """Coerce a value to float; anything non-numeric becomes 0.0."""
+    if isinstance(value, bool):
+        return 1.0 if value else 0.0
+    if isinstance(value, (int, float)):
+        return float(value)
+    if isinstance(value, str):
+        try:
+            return float(value.strip())
+        except ValueError:
+            return 0.0
+    return 0.0
+
+
+class FalJSONExtract:
+    """Pull a value out of a JSON result by dot/bracket path."""
+
+    RETURN_TYPES = ("STRING", "FLOAT", "BOOLEAN")
+    RETURN_NAMES = ("text", "number", "boolean")
+    FUNCTION = "extract"
+    CATEGORY = _CATEGORY
+    DESCRIPTION = (
+        "Extract a value from JSON text by path (e.g. video.url, "
+        "images[0].url). Returns it as text, number, and boolean so it can "
+        "wire straight into other nodes. Missing paths return the default "
+        "instead of failing the graph."
+    )
+
+    @classmethod
+    def INPUT_TYPES(cls) -> dict[str, Any]:
+        return {
+            "required": {
+                "json_text": (
+                    "STRING",
+                    {
+                        "forceInput": True,
+                        "multiline": True,
+                        "tooltip": (
+                            "Wire result_json from a Fal Any Endpoint / Fal Collect "
+                            "node here to pick values out of the raw API result."
+                        ),
+                    },
+                ),
+                "path": (
+                    "STRING",
+                    {
+                        "default": "video.url",
+                        "tooltip": (
+                            "Dot/bracket path into the JSON, e.g. video.url, "
+                            "images[0].url, data.items[2].name. Bare integers "
+                            "also index arrays (images.0.url)."
+                        ),
+                    },
+                ),
+                "default": (
+                    "STRING",
+                    {
+                        "default": "",
+                        "tooltip": "Returned as text when the path is missing (not an error)",
+                    },
+                ),
+            },
+        }
+
+    def extract(self, json_text: str, path: str = "video.url", default: str = "") -> tuple[str, float, bool]:
+        try:
+            payload = json.loads(json_text)
+        except Exception as exc:
+            logger.error("FalJSONExtract: invalid JSON input: %s", exc)
+            raise FalApiError("FalJSONExtract", f"Input is not valid JSON: {exc}") from exc
+
+        tokens = _tokenize_path(path or "")
+        value = _walk_path(payload, tokens) if tokens is not None else _MISSING
+        if value is _MISSING:
+            logger.debug("FalJSONExtract: path %r missing, returning default", path)
+            value = default
+        return (_value_to_text(value), _value_to_number(value), _value_to_bool(value))
+
+
+class FalPromptLines:
+    """Cycle through a multiline prompt list, one line per run."""
+
+    RETURN_TYPES = ("STRING", "INT", "INT")
+    RETURN_NAMES = ("line", "index", "total")
+    FUNCTION = "pick"
+    CATEGORY = _CATEGORY
+    DESCRIPTION = (
+        "Pick one line from a multiline text by index. The index wraps "
+        "around (modulo the number of lines), so with control_after_generate "
+        "set to increment it cycles through your prompt list forever."
+    )
+
+    @classmethod
+    def INPUT_TYPES(cls) -> dict[str, Any]:
+        return {
+            "required": {
+                "text": (
+                    "STRING",
+                    {
+                        "default": "",
+                        "multiline": True,
+                        "tooltip": "Prompt list, one prompt per line",
+                    },
+                ),
+                "index": (
+                    "INT",
+                    {
+                        "default": 0,
+                        "min": 0,
+                        "max": _MAX_INDEX,
+                        "control_after_generate": True,
+                        "tooltip": (
+                            "Which line to pick (wraps around). Set the control to "
+                            "'increment' to iterate through your prompt list run by run."
+                        ),
+                    },
+                ),
+                "skip_blank": (
+                    "BOOLEAN",
+                    {"default": True, "tooltip": "Ignore empty/whitespace-only lines"},
+                ),
+            },
+        }
+
+    def pick(self, text: str, index: int = 0, skip_blank: bool = True) -> tuple[str, int, int]:
+        lines = (text or "").splitlines()
+        if skip_blank:
+            lines = [line for line in lines if line.strip()]
+        total = len(lines)
+        if total == 0:
+            return ("", 0, 0)
+        effective = int(index) % total
+        return (lines[effective], effective, total)
+
+
+class FalTextTemplate:
+    """Fill a text template's {a}..{d} placeholders from string inputs."""
+
+    RETURN_TYPES = ("STRING",)
+    FUNCTION = "render"
+    CATEGORY = _CATEGORY
+    DESCRIPTION = (
+        "Substitute {a}, {b}, {c}, {d} placeholders in a template with the "
+        "connected string inputs — quick prompt assembly without string "
+        "concatenation chains. Missing inputs become empty text."
+    )
+
+    @classmethod
+    def INPUT_TYPES(cls) -> dict[str, Any]:
+        return {
+            "required": {
+                "template": (
+                    "STRING",
+                    {
+                        "default": "a photo of {a}, {b} style",
+                        "multiline": True,
+                        "tooltip": "Template text; {a} {b} {c} {d} are replaced with the inputs below",
+                    },
+                ),
+            },
+            "optional": {
+                "a": ("STRING", {"default": "", "tooltip": "Value for {a}"}),
+                "b": ("STRING", {"default": "", "tooltip": "Value for {b}"}),
+                "c": ("STRING", {"default": "", "tooltip": "Value for {c}"}),
+                "d": ("STRING", {"default": "", "tooltip": "Value for {d}"}),
+            },
+        }
+
+    def render(self, template: str, a: str = "", b: str = "", c: str = "", d: str = "") -> tuple[str]:
+        result = template or ""
+        for key, value in (("a", a), ("b", b), ("c", c), ("d", d)):
+            result = result.replace("{" + key + "}", value or "")
+        return (result,)
+
+
+NODE_CLASS_MAPPINGS = {
+    "FalJSONExtract_fal": FalJSONExtract,
+    "FalPromptLines_fal": FalPromptLines,
+    "FalTextTemplate_fal": FalTextTemplate,
+}
+
+NODE_DISPLAY_NAME_MAPPINGS = {
+    "FalJSONExtract_fal": "JSON Extract (fal)",
+    "FalPromptLines_fal": "Prompt Lines (fal)",
+    "FalTextTemplate_fal": "Text Template (fal)",
+}

nodes/util_dataset_node.py

@@ -0,0 +1,409 @@
+"""Dataset preparation utility nodes (zip building, frame extraction, captioning)."""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import zipfile
+from concurrent.futures import ThreadPoolExecutor
+from typing import Any
+
+from .fal_utils import (
+    ApiHandler,
+    ArchiveUtils,
+    FalApiError,
+    FalConfig,
+    ImageUtils,
+    MediaUtils,
+    logger,
+)
+
+# Initialize FalConfig
+fal_config = FalConfig()
+
+_CATEGORY = "FAL/Utils/Dataset"
+_ARCHIVE_MODEL = "archive"
+_VISION_ENDPOINT = "openrouter/router/vision"
+_MAX_CAPTION_WORKERS = 8
+_STREAM_CHUNK_SIZE = 1 << 20  # 1 MiB
+
+
+def _safe_unlink(path: str | None) -> None:
+    """Delete a temp file, ignoring errors."""
+    if path is None:
+        return
+    try:
+        os.unlink(path)
+    except OSError:
+        pass
+
+
+def _split_caption_lines(captions: str) -> list[str] | None:
+    """Split a multiline caption field into one caption per line (None if empty)."""
+    if not captions or not captions.strip():
+        return None
+    return captions.splitlines()
+
+
+def _video_to_local_path(video: Any) -> tuple[str, bool]:
+    """Resolve a VIDEO input to a local file path. Returns (path, is_temp)."""
+    source = video.get_stream_source() if hasattr(video, "get_stream_source") else video
+    if isinstance(source, str):
+        if source.startswith(("http://", "https://")):
+            return MediaUtils.download_url_to_temp(source, ".mp4"), True
+        if not os.path.isfile(source):
+            raise FalApiError(
+                _ARCHIVE_MODEL,
+                f"Video file not found: {source}. Connect a valid VIDEO input.",
+            )
+        return source, False
+    if hasattr(source, "read"):
+        temp_path: str | None = None
+        try:
+            with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as temp_file:
+                temp_path = temp_file.name
+                while True:
+                    chunk = source.read(_STREAM_CHUNK_SIZE)
+                    if not chunk:
+                        break
+                    temp_file.write(chunk)
+            return temp_path, True
+        except Exception as exc:
+            _safe_unlink(temp_path)
+            raise FalApiError(
+                _ARCHIVE_MODEL, f"Failed to buffer video stream to disk: {exc}"
+            ) from exc
+    raise FalApiError(
+        _ARCHIVE_MODEL,
+        "Unsupported VIDEO input: could not resolve a local file, URL, or stream from it.",
+    )
+
+
+def _extract_frames_to_zip(video_path: str, every_nth: int, max_frames: int) -> str:
+    """Decode a video with cv2, sample every Nth frame as PNG into a zip, return the zip path."""
+    try:
+        import cv2
+    except ImportError as exc:
+        raise FalApiError(
+            _ARCHIVE_MODEL,
+            "opencv-python is required to extract video frames. "
+            "Install it with 'pip install opencv-python'.",
+        ) from exc
+
+    capture = cv2.VideoCapture(video_path)
+    if not capture.isOpened():
+        raise FalApiError(
+            _ARCHIVE_MODEL,
+            f"Could not open video for decoding: {video_path}. "
+            "Check that the input is a valid video file.",
+        )
+
+    zip_path: str | None = None
+    try:
+        with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as temp_zip:
+            zip_path = temp_zip.name
+        saved = 0
+        index = 0
+        with zipfile.ZipFile(zip_path, "w") as zip_file:
+            while saved < max_frames:
+                ok, frame = capture.read()
+                if not ok:
+                    break
+                if index % every_nth == 0:
+                    encoded, buffer = cv2.imencode(".png", frame)
+                    if not encoded:
+                        raise FalApiError(
+                            _ARCHIVE_MODEL, f"Failed to encode frame {index} as PNG."
+                        )
+                    zip_file.writestr(f"frame_{saved:05d}.png", buffer.tobytes())
+                    saved += 1
+                index += 1
+        if saved == 0:
+            raise FalApiError(
+                _ARCHIVE_MODEL,
+                "No frames could be decoded from the video. "
+                "Check the input video and the every_nth setting.",
+            )
+        logger.info("Extracted %d frame(s) from %s", saved, video_path)
+        return zip_path
+    except Exception:
+        _safe_unlink(zip_path)
+        raise
+    finally:
+        capture.release()
+
+
+class FalImagesToZipURL:
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "images": (
+                    "IMAGE",
+                    {
+                        "tooltip": "Images to package as a training dataset zip (image_0.png, image_1.png, ...).",
+                    },
+                ),
+            },
+            "optional": {
+                "captions": (
+                    "STRING",
+                    {
+                        "default": "",
+                        "multiline": True,
+                        "tooltip": "Optional captions, one per line (blank lines allowed). "
+                        "Line count must match the image batch size, or leave empty for no captions. "
+                        "Written as image_0.txt, image_1.txt, ... next to each image.",
+                    },
+                ),
+                "name_prefix": (
+                    "STRING",
+                    {
+                        "default": "image",
+                        "tooltip": "File name prefix inside the zip (e.g. 'image' -> image_0.png).",
+                    },
+                ),
+            },
+        }
+
+    RETURN_TYPES = ("STRING",)
+    RETURN_NAMES = ("zip_url",)
+    FUNCTION = "create_zip_url"
+    CATEGORY = _CATEGORY
+    DESCRIPTION = (
+        "Zips an IMAGE batch (with optional per-image captions) and uploads it to fal.ai. "
+        "Feed the URL directly into the LoRA trainer nodes' images_data_url input."
+    )
+
+    def create_zip_url(self, images, captions="", name_prefix="image"):
+        caption_lines = _split_caption_lines(captions)
+        zip_path = ArchiveUtils.zip_images(
+            images, captions=caption_lines, name_prefix=name_prefix
+        )
+        return (ArchiveUtils.upload_zip(zip_path),)
+
+
+class FalFolderToZipURL:
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "folder_path": (
+                    "STRING",
+                    {
+                        "default": "",
+                        "tooltip": "Path to a local folder whose files will be zipped and uploaded.",
+                    },
+                ),
+                "recursive": (
+                    "BOOLEAN",
+                    {
+                        "default": False,
+                        "tooltip": "Also include files from subfolders (hidden entries are always skipped).",
+                    },
+                ),
+                "extensions": (
+                    "STRING",
+                    {
+                        "default": ".png,.jpg,.jpeg,.webp,.txt",
+                        "tooltip": "Comma-separated list of file extensions to include. Empty includes all files.",
+                    },
+                ),
+            },
+        }
+
+    RETURN_TYPES = ("STRING",)
+    RETURN_NAMES = ("zip_url",)
+    FUNCTION = "create_zip_url"
+    CATEGORY = _CATEGORY
+    DESCRIPTION = "Zips a local folder and uploads it to fal.ai, returning the zip URL."
+
+    def create_zip_url(self, folder_path, recursive=False, extensions=".png,.jpg,.jpeg,.webp,.txt"):
+        extension_list = [part.strip() for part in extensions.split(",") if part.strip()]
+        zip_path = ArchiveUtils.zip_folder(
+            folder_path,
+            include_extensions=extension_list or None,
+            recursive=recursive,
+        )
+        return (ArchiveUtils.upload_zip(zip_path),)
+
+
+class FalVideoToFrameDatasetZip:
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "video": (
+                    "VIDEO",
+                    {
+                        "tooltip": "Video to sample frames from for a training dataset.",
+                    },
+                ),
+                "every_nth": (
+                    "INT",
+                    {
+                        "default": 10,
+                        "min": 1,
+                        "max": 10000,
+                        "step": 1,
+                        "tooltip": "Keep one frame out of every N decoded frames.",
+                    },
+                ),
+                "max_frames": (
+                    "INT",
+                    {
+                        "default": 200,
+                        "min": 1,
+                        "max": 2000,
+                        "step": 1,
+                        "tooltip": "Stop after this many frames have been saved.",
+                    },
+                ),
+            },
+        }
+
+    RETURN_TYPES = ("STRING",)
+    RETURN_NAMES = ("zip_url",)
+    FUNCTION = "create_zip_url"
+    CATEGORY = _CATEGORY
+    DESCRIPTION = (
+        "Samples frames from a video (every Nth frame, up to max_frames), zips them as PNGs, "
+        "uploads the zip to fal.ai, and returns the URL."
+    )
+
+    def create_zip_url(self, video, every_nth=10, max_frames=200):
+        local_path, is_temp = _video_to_local_path(video)
+        try:
+            zip_path = _extract_frames_to_zip(local_path, every_nth, max_frames)
+        finally:
+            if is_temp:
+                _safe_unlink(local_path)
+        return (ArchiveUtils.upload_zip(zip_path),)
+
+
+class FalBatchCaption:
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "images": (
+                    "IMAGE",
+                    {
+                        "tooltip": "Images to caption, one caption per frame.",
+                    },
+                ),
+                "prompt": (
+                    "STRING",
+                    {
+                        "default": "Describe this image for LoRA training in one dense sentence.",
+                        "multiline": True,
+                        "tooltip": "Instruction sent to the vision model for each image.",
+                    },
+                ),
+                "model": (
+                    [
+                        "google/gemini-2.5-flash",
+                        "anthropic/claude-sonnet-4.5",
+                        "openai/gpt-4o",
+                        "custom",
+                    ],
+                    {
+                        "default": "google/gemini-2.5-flash",
+                        "tooltip": "Vision model to use. Select 'custom' to type any OpenRouter model id "
+                        "in custom_model_name.",
+                    },
+                ),
+            },
+            "optional": {
+                "custom_model_name": (
+                    "STRING",
+                    {
+                        "default": "",
+                        "tooltip": "OpenRouter model id used when model is set to 'custom'.",
+                    },
+                ),
+            },
+        }
+
+    RETURN_TYPES = ("STRING",)
+    RETURN_NAMES = ("captions",)
+    FUNCTION = "caption_images"
+    CATEGORY = _CATEGORY
+    DESCRIPTION = (
+        "Captions each image in a batch with a fal VLM (concurrently, order preserved) and returns "
+        "one caption per line — wire straight into FalImagesToZipURL's captions input."
+    )
+
+    def caption_images(
+        self,
+        images,
+        prompt="Describe this image for LoRA training in one dense sentence.",
+        model="google/gemini-2.5-flash",
+        custom_model_name="",
+    ):
+        if model == "custom":
+            if not custom_model_name or not custom_model_name.strip():
+                raise FalApiError(
+                    _VISION_ENDPOINT,
+                    "custom_model_name is required when model is set to 'custom'.",
+                )
+            model = custom_model_name.strip()
+
+        image_urls = ImageUtils.prepare_images(images)
+        if not image_urls:
+            raise FalApiError(
+                _VISION_ENDPOINT, "No images provided to caption. Connect an IMAGE batch."
+            )
+
+        def caption_one(image_url: str) -> str:
+            arguments = {
+                "model": model,
+                "prompt": prompt,
+                "image_urls": [image_url],
+                "stream": False,
+            }
+            result = ApiHandler.submit_and_get_result(_VISION_ENDPOINT, arguments)
+            # Captions are joined by newline, so flatten any multiline output.
+            return str(result["output"]).replace("\r", " ").replace("\n", " ").strip()
+
+        with ThreadPoolExecutor(max_workers=_MAX_CAPTION_WORKERS) as executor:
+            futures = [executor.submit(caption_one, url) for url in image_urls]
+
+        captions: list[str] = []
+        failure_count = 0
+        for index, future in enumerate(futures):
+            try:
+                captions = [*captions, future.result()]
+            except Exception as exc:
+                # a user Cancel raised inside a worker must stop the node,
+                # not silently become an empty caption
+                if exc.__class__.__name__ == "InterruptProcessingException":
+                    raise
+                logger.warning("Caption for image %d failed: %s", index, exc)
+                captions = [*captions, ""]
+                failure_count += 1
+
+        if failure_count == len(image_urls):
+            raise FalApiError(
+                _VISION_ENDPOINT,
+                f"All {len(image_urls)} caption request(s) failed. "
+                "Check the model id, your fal API key, and the queue logs above.",
+            )
+        return ("\n".join(captions),)
+
+
+# Node class mappings
+NODE_CLASS_MAPPINGS = {
+    "FalImagesToZipURL_fal": FalImagesToZipURL,
+    "FalFolderToZipURL_fal": FalFolderToZipURL,
+    "FalVideoToFrameDatasetZip_fal": FalVideoToFrameDatasetZip,
+    "FalBatchCaption_fal": FalBatchCaption,
+}
+
+# Node display name mappings
+NODE_DISPLAY_NAME_MAPPINGS = {
+    "FalImagesToZipURL_fal": "Images → Training ZIP URL (fal)",
+    "FalFolderToZipURL_fal": "Folder → ZIP URL (fal)",
+    "FalVideoToFrameDatasetZip_fal": "Video → Frame Dataset ZIP URL (fal)",
+    "FalBatchCaption_fal": "Batch Caption Images (fal VLM)",
+}

nodes/util_image_node.py

@@ -0,0 +1,423 @@
+"""Image utility nodes: labeled grids, preset resizing, and base64 conversion."""
+
+from __future__ import annotations
+
+import base64
+import io
+import math
+import re
+from typing import Any
+
+import numpy as np
+import torch
+from PIL import Image, ImageDraw, ImageFont
+
+from .fal_utils import FalApiError, ImageUtils, logger
+
+_CATEGORY = "FAL/Utils/Image"
+
+# fal image_size preset -> (width, height)
+_PRESET_SIZES = {
+    "square_hd": (1024, 1024),
+    "square": (512, 512),
+    "portrait_4_3": (768, 1024),
+    "portrait_16_9": (576, 1024),
+    "landscape_4_3": (1024, 768),
+    "landscape_16_9": (1024, 576),
+}
+_CUSTOM_PRESET = "custom"
+
+_LANCZOS = getattr(Image, "Resampling", Image).LANCZOS
+
+_GRID_BG = (24, 24, 24)
+_LABEL_BG = (16, 16, 16)
+_LABEL_FG = (235, 235, 235)
+_LABEL_MARGIN = 4
+_ELLIPSIS = "..."
+
+_B64_WHITESPACE = re.compile(r"\s+")
+_MIME_BY_FORMAT = {"png": "image/png", "jpeg": "image/jpeg", "webp": "image/webp"}
+
+
+def _pils_to_tensor(pils: list[Image.Image]) -> torch.Tensor:
+    """Stack same-sized PIL images into a float32 (B, H, W, 3) IMAGE tensor."""
+    arrays = [np.array(pil.convert("RGB")).astype(np.float32) / 255.0 for pil in pils]
+    return torch.from_numpy(np.stack(arrays, axis=0))
+
+
+def _image_input_to_pils(images: Any) -> list[Image.Image]:
+    """Convert an IMAGE input (batch tensor or list of tensors) to PIL images."""
+    if isinstance(images, torch.Tensor) and images.ndim == 4:
+        items: list[Any] = [images[i] for i in range(images.shape[0])]
+    elif isinstance(images, (list, tuple)):
+        items = list(images)
+    else:
+        items = [images]
+    if not items:
+        raise FalApiError("FalImageGrid", "IMAGE input contained no images")
+    return [ImageUtils.tensor_to_pil(item) for item in items]
+
+
+def _letterbox(pil: Image.Image, width: int, height: int, fill: tuple[int, int, int]) -> Image.Image:
+    """Fit an image inside (width, height) preserving aspect, padded with fill."""
+    scale = min(width / pil.width, height / pil.height)
+    new_size = (max(1, round(pil.width * scale)), max(1, round(pil.height * scale)))
+    resized = pil.convert("RGB").resize(new_size, _LANCZOS)
+    canvas = Image.new("RGB", (width, height), fill)
+    offset = ((width - new_size[0]) // 2, (height - new_size[1]) // 2)
+    canvas.paste(resized, offset)
+    return canvas
+
+
+def _truncate_label(draw: ImageDraw.ImageDraw, text: str, font: Any, max_width: int) -> str:
+    """Truncate text with an ellipsis so it fits within max_width pixels."""
+    if draw.textlength(text, font=font) <= max_width:
+        return text
+    for end in range(len(text) - 1, 0, -1):
+        candidate = text[:end].rstrip() + _ELLIPSIS
+        if draw.textlength(candidate, font=font) <= max_width:
+            return candidate
+    return _ELLIPSIS
+
+
+def _draw_label(
+    canvas: Image.Image, text: str, x: int, y: int, cell_width: int, label_height: int
+) -> None:
+    """Draw one centered label line on its dark strip below a cell."""
+    draw = ImageDraw.Draw(canvas)
+    draw.rectangle((x, y, x + cell_width - 1, y + label_height - 1), fill=_LABEL_BG)
+    if not text:
+        return
+    font = ImageFont.load_default()
+    fitted = _truncate_label(draw, text, font, cell_width - 2 * _LABEL_MARGIN)
+    text_width = draw.textlength(fitted, font=font)
+    bbox = font.getbbox(fitted)
+    text_height = bbox[3] - bbox[1]
+    text_x = x + max(_LABEL_MARGIN, (cell_width - text_width) // 2)
+    text_y = y + max(0, (label_height - text_height) // 2) - bbox[1]
+    draw.text((text_x, text_y), fitted, font=font, fill=_LABEL_FG)
+
+
+def _grid_shape(count: int, columns: int) -> tuple[int, int]:
+    """Resolve (columns, rows) for a grid; columns == 0 means auto square-ish."""
+    cols = columns if columns > 0 else math.ceil(math.sqrt(count))
+    cols = max(1, min(cols, count))
+    return cols, math.ceil(count / cols)
+
+
+def _compose_grid(
+    pils: list[Image.Image], labels: list[str], columns: int, padding: int, label_height: int
+) -> Image.Image:
+    """Lay out letterboxed cells (plus optional label strips) on a dark canvas."""
+    cell_w = max(pil.width for pil in pils)
+    cell_h = max(pil.height for pil in pils)
+    strip_h = label_height if labels else 0
+    cols, rows = _grid_shape(len(pils), columns)
+    total_w = cols * cell_w + (cols + 1) * padding
+    total_h = rows * (cell_h + strip_h) + (rows + 1) * padding
+    canvas = Image.new("RGB", (total_w, total_h), _GRID_BG)
+    for i, pil in enumerate(pils):
+        col, row = i % cols, i // cols
+        x = padding + col * (cell_w + padding)
+        y = padding + row * (cell_h + strip_h + padding)
+        canvas.paste(_letterbox(pil, cell_w, cell_h, _GRID_BG), (x, y))
+        if strip_h:
+            text = labels[i] if i < len(labels) else ""
+            _draw_label(canvas, text, x, y + cell_h, cell_w, strip_h)
+    return canvas
+
+
+class FalImageGrid:
+    """Compose an image batch into a single labeled contact-sheet grid."""
+
+    RETURN_TYPES = ("IMAGE",)
+    FUNCTION = "compose"
+    CATEGORY = _CATEGORY
+    DESCRIPTION = (
+        "Arrange a batch of images into one grid image with optional text "
+        "labels under each cell. Mixed sizes are letterboxed into uniform "
+        "cells on a dark background — handy for comparing seeds, prompts, "
+        "or models side by side."
+    )
+
+    @classmethod
+    def INPUT_TYPES(cls) -> dict[str, Any]:
+        return {
+            "required": {
+                "images": ("IMAGE", {"tooltip": "Batch of images to arrange into a grid"}),
+            },
+            "optional": {
+                "labels": (
+                    "STRING",
+                    {
+                        "default": "",
+                        "multiline": True,
+                        "tooltip": (
+                            "One label per line, matched to images in batch order. "
+                            "Leave empty for no label strips."
+                        ),
+                    },
+                ),
+                "columns": (
+                    "INT",
+                    {
+                        "default": 0,
+                        "min": 0,
+                        "max": 64,
+                        "tooltip": "Number of grid columns; 0 = auto (roughly square)",
+                    },
+                ),
+                "cell_padding": (
+                    "INT",
+                    {
+                        "default": 8,
+                        "min": 0,
+                        "max": 64,
+                        "tooltip": "Pixels of dark padding around each cell",
+                    },
+                ),
+                "label_height": (
+                    "INT",
+                    {
+                        "default": 28,
+                        "min": 12,
+                        "max": 128,
+                        "tooltip": "Height in pixels of the label strip under each cell",
+                    },
+                ),
+            },
+        }
+
+    def compose(
+        self,
+        images: Any,
+        labels: str = "",
+        columns: int = 0,
+        cell_padding: int = 8,
+        label_height: int = 28,
+    ) -> tuple[torch.Tensor]:
+        pils = _image_input_to_pils(images)
+        label_lines = [line.strip() for line in labels.splitlines()] if labels.strip() else []
+        try:
+            grid = _compose_grid(pils, label_lines, int(columns), int(cell_padding), int(label_height))
+        except FalApiError:
+            raise
+        except Exception as exc:
+            logger.error("FalImageGrid: failed to compose grid: %s", exc)
+            raise FalApiError("FalImageGrid", f"Failed to compose image grid: {exc}") from exc
+        logger.debug("FalImageGrid: composed %d cells into %dx%d", len(pils), grid.width, grid.height)
+        return (_pils_to_tensor([grid]),)
+
+
+def _resize_one(pil: Image.Image, width: int, height: int, mode: str) -> Image.Image:
+    """Resize a single PIL image to (width, height) using the given mode."""
+    source = pil.convert("RGB")
+    if mode == "stretch":
+        return source.resize((width, height), _LANCZOS)
+    if mode == "contain_pad":
+        return _letterbox(source, width, height, (0, 0, 0))
+    # cover_crop: scale to fully cover the target, then center-crop
+    scale = max(width / source.width, height / source.height)
+    scaled = source.resize(
+        (max(width, round(source.width * scale)), max(height, round(source.height * scale))),
+        _LANCZOS,
+    )
+    left = (scaled.width - width) // 2
+    top = (scaled.height - height) // 2
+    return scaled.crop((left, top, left + width, top + height))
+
+
+class FalResizeToPreset:
+    """Resize images to an exact fal image_size preset (or custom dimensions)."""
+
+    RETURN_TYPES = ("IMAGE", "INT", "INT")
+    RETURN_NAMES = ("image", "width", "height")
+    FUNCTION = "resize"
+    CATEGORY = _CATEGORY
+    DESCRIPTION = (
+        "Resize images to the exact pixel dimensions of a fal image_size "
+        "preset (square_hd, portrait_16_9, ...) or custom width/height. "
+        "Choose cover (crop), contain (letterbox), or stretch. Batch-safe."
+    )
+
+    @classmethod
+    def INPUT_TYPES(cls) -> dict[str, Any]:
+        return {
+            "required": {
+                "image": ("IMAGE", {"tooltip": "Image (or batch) to resize"}),
+                "preset": (
+                    [*_PRESET_SIZES.keys(), _CUSTOM_PRESET],
+                    {
+                        "default": "square_hd",
+                        "tooltip": (
+                            "fal image_size preset: square_hd=1024x1024, square=512x512, "
+                            "portrait_4_3=768x1024, portrait_16_9=576x1024, "
+                            "landscape_4_3=1024x768, landscape_16_9=1024x576. "
+                            "'custom' uses the width/height inputs."
+                        ),
+                    },
+                ),
+                "width": (
+                    "INT",
+                    {
+                        "default": 1024,
+                        "min": 8,
+                        "max": 14142,
+                        "step": 8,
+                        "tooltip": "Target width in pixels (used when preset is 'custom')",
+                    },
+                ),
+                "height": (
+                    "INT",
+                    {
+                        "default": 1024,
+                        "min": 8,
+                        "max": 14142,
+                        "step": 8,
+                        "tooltip": "Target height in pixels (used when preset is 'custom')",
+                    },
+                ),
+                "mode": (
+                    ["cover_crop", "contain_pad", "stretch"],
+                    {
+                        "default": "cover_crop",
+                        "tooltip": (
+                            "cover_crop: fill the frame and center-crop the overflow; "
+                            "contain_pad: fit inside and letterbox with black bars; "
+                            "stretch: ignore aspect ratio"
+                        ),
+                    },
+                ),
+            },
+        }
+
+    def resize(
+        self,
+        image: Any,
+        preset: str = "square_hd",
+        width: int = 1024,
+        height: int = 1024,
+        mode: str = "cover_crop",
+    ) -> tuple[torch.Tensor, int, int]:
+        if preset == _CUSTOM_PRESET:
+            target_w, target_h = int(width), int(height)
+        elif preset in _PRESET_SIZES:
+            target_w, target_h = _PRESET_SIZES[preset]
+        else:
+            raise FalApiError("FalResizeToPreset", f"Unknown preset: {preset!r}")
+        if target_w < 1 or target_h < 1:
+            raise FalApiError("FalResizeToPreset", f"Invalid target size: {target_w}x{target_h}")
+
+        pils = _image_input_to_pils(image)
+        try:
+            resized = [_resize_one(pil, target_w, target_h, mode) for pil in pils]
+        except Exception as exc:
+            logger.error("FalResizeToPreset: resize failed: %s", exc)
+            raise FalApiError("FalResizeToPreset", f"Failed to resize image: {exc}") from exc
+        return (_pils_to_tensor(resized), target_w, target_h)
+
+
+class FalImageToBase64:
+    """Encode an image as a base64 string (optionally a data: URI)."""
+
+    RETURN_TYPES = ("STRING",)
+    FUNCTION = "encode"
+    CATEGORY = _CATEGORY
+    DESCRIPTION = (
+        "Encode the first image of a batch as base64 text, optionally "
+        "wrapped in a data: URI — useful for APIs that accept inline "
+        "base64 images instead of URLs."
+    )
+
+    @classmethod
+    def INPUT_TYPES(cls) -> dict[str, Any]:
+        return {
+            "required": {
+                "image": ("IMAGE", {"tooltip": "Image to encode (first of the batch is used)"}),
+                "format": (
+                    ["png", "jpeg", "webp"],
+                    {"default": "png", "tooltip": "Encoding format; png and webp are lossless-capable"},
+                ),
+                "data_uri": (
+                    "BOOLEAN",
+                    {
+                        "default": True,
+                        "tooltip": "Prefix with 'data:image/...;base64,' (most APIs expect this)",
+                    },
+                ),
+            },
+        }
+
+    def encode(self, image: Any, format: str = "png", data_uri: bool = True) -> tuple[str]:
+        fmt = (format or "png").lower()
+        if fmt not in _MIME_BY_FORMAT:
+            raise FalApiError("FalImageToBase64", f"Unsupported format: {format!r}")
+        pil = ImageUtils.tensor_to_pil(image).convert("RGB")
+        try:
+            buffer = io.BytesIO()
+            save_kwargs = {"lossless": True} if fmt == "webp" else {}
+            pil.save(buffer, format=fmt.upper(), **save_kwargs)
+            encoded = base64.b64encode(buffer.getvalue()).decode("ascii")
+        except Exception as exc:
+            logger.error("FalImageToBase64: encoding failed: %s", exc)
+            raise FalApiError("FalImageToBase64", f"Failed to encode image as {fmt}: {exc}") from exc
+        if data_uri:
+            return (f"data:{_MIME_BY_FORMAT[fmt]};base64,{encoded}",)
+        return (encoded,)
+
+
+class FalBase64ToImage:
+    """Decode a base64 string (raw or data: URI) into an IMAGE tensor."""
+
+    RETURN_TYPES = ("IMAGE",)
+    FUNCTION = "decode"
+    CATEGORY = _CATEGORY
+    DESCRIPTION = (
+        "Decode base64 image data — either a raw base64 string or a full "
+        "'data:image/...;base64,...' URI — into a ComfyUI IMAGE."
+    )
+
+    @classmethod
+    def INPUT_TYPES(cls) -> dict[str, Any]:
+        return {
+            "required": {
+                "data": (
+                    "STRING",
+                    {
+                        "default": "",
+                        "multiline": True,
+                        "tooltip": "Raw base64 image data, or a data:image/...;base64,... URI",
+                    },
+                ),
+            },
+        }
+
+    def decode(self, data: str) -> tuple[torch.Tensor]:
+        payload = (data or "").strip()
+        if payload.startswith("data:"):
+            _, _, payload = payload.partition(",")
+        payload = _B64_WHITESPACE.sub("", payload)
+        if not payload:
+            raise FalApiError("FalBase64ToImage", "No base64 data provided")
+        try:
+            raw = base64.b64decode(payload)
+            pil = Image.open(io.BytesIO(raw)).convert("RGB")
+        except Exception as exc:
+            logger.error("FalBase64ToImage: decoding failed: %s", exc)
+            raise FalApiError("FalBase64ToImage", f"Failed to decode base64 image: {exc}") from exc
+        return (_pils_to_tensor([pil]),)
+
+
+NODE_CLASS_MAPPINGS = {
+    "FalImageGrid_fal": FalImageGrid,
+    "FalResizeToPreset_fal": FalResizeToPreset,
+    "FalImageToBase64_fal": FalImageToBase64,
+    "FalBase64ToImage_fal": FalBase64ToImage,
+}
+
+NODE_DISPLAY_NAME_MAPPINGS = {
+    "FalImageGrid_fal": "Image Grid with Labels (fal)",
+    "FalResizeToPreset_fal": "Resize to fal Preset (fal)",
+    "FalImageToBase64_fal": "Image → Base64 (fal)",
+    "FalBase64ToImage_fal": "Base64 → Image (fal)",
+}

nodes/util_media_in_node.py

@@ -0,0 +1,404 @@
+"""Utility loader nodes: bring images/audio/folders into ComfyUI from URLs and disk.
+
+ComfyUI IMAGE convention: float32 tensors in [0, 1] with shape (B, H, W, C).
+"""
+
+from __future__ import annotations
+
+import glob
+import io
+import os
+from typing import Any
+
+import numpy as np
+import requests
+import torch
+from PIL import Image
+
+from .fal_utils import FalApiError, MediaUtils, logger
+
+_CATEGORY = "FAL/Utils/Load"
+_DOWNLOAD_TIMEOUT = (10, 180)
+_DEFAULT_FOLDER_PATTERN = "*.png,*.jpg,*.jpeg,*.webp"
+
+
+def _split_csv(value: str) -> list[str]:
+    """Split a comma-separated string into stripped, non-empty parts."""
+    return [part.strip() for part in (value or "").split(",") if part.strip()]
+
+
+def _validate_http_url(node_name: str, url: str) -> str:
+    """Validate that a URL is a non-empty http(s) URL and return it stripped."""
+    stripped = (url or "").strip()
+    if not stripped:
+        raise FalApiError(
+            node_name, "'url' is empty. Provide an http(s) URL to a media file."
+        )
+    if not stripped.startswith(("http://", "https://")):
+        raise FalApiError(
+            node_name,
+            f"Invalid URL '{stripped}'. Only http(s) URLs are supported.",
+        )
+    return stripped
+
+
+def _download_pil_image(node_name: str, url: str) -> Image.Image:
+    """Download a URL and decode it as an RGB PIL image."""
+    try:
+        response = requests.get(url, timeout=_DOWNLOAD_TIMEOUT)
+        response.raise_for_status()
+        return Image.open(io.BytesIO(response.content)).convert("RGB")
+    except FalApiError:
+        raise
+    except Exception as exc:
+        logger.error("%s: failed to download image %s: %s", node_name, url, exc)
+        raise FalApiError(
+            node_name,
+            f"Failed to download or decode image from '{url}': {exc}",
+        ) from exc
+
+
+def _images_to_batch_tensor(
+    node_name: str, images: list[Image.Image], labels: list[str]
+) -> torch.Tensor:
+    """Stack RGB PIL images into a float32 (B, H, W, C) tensor in [0, 1].
+
+    Images whose size differs from the first image are resized to match
+    (with a warning) so the batch stays valid.
+    """
+    first_size = images[0].size  # (W, H)
+    arrays: list[np.ndarray] = []
+    for img, label in zip(images, labels):
+        if img.size != first_size:
+            logger.warning(
+                "%s: '%s' is %sx%s; resizing to %sx%s to match the first image",
+                node_name,
+                label,
+                img.size[0],
+                img.size[1],
+                first_size[0],
+                first_size[1],
+            )
+            img = img.resize(first_size, Image.LANCZOS)
+        arrays.append(np.array(img).astype(np.float32) / 255.0)
+    return torch.from_numpy(np.stack(arrays, axis=0))
+
+
+class FalLoadImageURL:
+    """Load one or more images from http(s) URLs into an IMAGE batch."""
+
+    RETURN_TYPES = ("IMAGE",)
+    RETURN_NAMES = ("image",)
+    FUNCTION = "load"
+    CATEGORY = _CATEGORY
+    DESCRIPTION = (
+        "Download an image from an http(s) URL into a ComfyUI IMAGE tensor. "
+        "Accepts a comma-separated list of URLs to build a batch; images with "
+        "differing sizes are resized to match the first."
+    )
+
+    @classmethod
+    def INPUT_TYPES(cls) -> dict[str, Any]:
+        return {
+            "required": {
+                "url": (
+                    "STRING",
+                    {
+                        "default": "",
+                        "multiline": True,
+                        "tooltip": (
+                            "http(s) URL of the image to load. A comma-separated "
+                            "list of URLs produces a batched IMAGE; mismatched "
+                            "sizes are resized to the first image's size. URLs "
+                            "containing literal commas are not supported in list mode."
+                        ),
+                    },
+                ),
+            },
+        }
+
+    def load(self, url: str) -> tuple[torch.Tensor]:
+        node_name = "FalLoadImageURL"
+        urls = [_validate_http_url(node_name, part) for part in _split_csv(url)]
+        if not urls:
+            raise FalApiError(
+                node_name,
+                "'url' is empty. Provide an http(s) URL (or a comma-separated "
+                "list of URLs) to image file(s).",
+            )
+        images = [_download_pil_image(node_name, u) for u in urls]
+        return (_images_to_batch_tensor(node_name, images, urls),)
+
+
+class FalLoadAudioURL:
+    """Load audio from an http(s) URL into a ComfyUI AUDIO output."""
+
+    RETURN_TYPES = ("AUDIO",)
+    RETURN_NAMES = ("audio",)
+    FUNCTION = "load"
+    CATEGORY = _CATEGORY
+    DESCRIPTION = (
+        "Download and decode an audio file from an http(s) URL into a native "
+        "ComfyUI AUDIO output (waveform + sample rate)."
+    )
+
+    @classmethod
+    def INPUT_TYPES(cls) -> dict[str, Any]:
+        return {
+            "required": {
+                "url": (
+                    "STRING",
+                    {
+                        "default": "",
+                        "tooltip": (
+                            "http(s) URL of the audio file to download and "
+                            "decode (e.g. the audio_url output of a fal node)."
+                        ),
+                    },
+                ),
+            },
+        }
+
+    def load(self, url: str) -> tuple[dict[str, Any]]:
+        node_name = "FalLoadAudioURL"
+        validated = _validate_http_url(node_name, url)
+        try:
+            return (MediaUtils.audio_from_url(validated),)
+        except FalApiError:
+            raise
+        except Exception as exc:
+            logger.error("%s: failed to load audio %s: %s", node_name, validated, exc)
+            raise FalApiError(
+                node_name,
+                f"Failed to load audio from '{validated}': {exc}",
+            ) from exc
+
+
+def _resolve_folder(node_name: str, folder_path: str) -> str:
+    """Expand and validate a folder path, returning its absolute form."""
+    expanded = os.path.expanduser((folder_path or "").strip())
+    if not expanded:
+        raise FalApiError(
+            node_name, "'folder_path' is empty. Provide a path to a folder."
+        )
+    if not os.path.isdir(expanded):
+        raise FalApiError(
+            node_name,
+            f"Folder not found: '{expanded}'. Provide an existing folder path.",
+        )
+    return os.path.abspath(expanded)
+
+
+def _glob_folder_files(folder: str, patterns: list[str]) -> list[str]:
+    """Glob a folder with each pattern, deduplicated, unordered."""
+    matched: set[str] = set()
+    for pattern in patterns:
+        for path in glob.glob(os.path.join(folder, pattern)):
+            if os.path.isfile(path):
+                matched.add(os.path.abspath(path))
+    return list(matched)
+
+
+def _sort_files(files: list[str], sort: str) -> list[str]:
+    """Sort file paths deterministically by name or modification time."""
+    if sort == "modified":
+        return sorted(files, key=lambda path: (os.path.getmtime(path), path))
+    return sorted(files)
+
+
+class FalLoadImageFolder:
+    """Load a folder of images from disk into a single IMAGE batch."""
+
+    RETURN_TYPES = ("IMAGE", "INT")
+    RETURN_NAMES = ("images", "count")
+    FUNCTION = "load"
+    CATEGORY = _CATEGORY
+    DESCRIPTION = (
+        "Load every image matching the pattern(s) in a local folder into one "
+        "IMAGE batch. Mixed sizes are resized to the first image's dimensions."
+    )
+
+    @classmethod
+    def INPUT_TYPES(cls) -> dict[str, Any]:
+        return {
+            "required": {
+                "folder_path": (
+                    "STRING",
+                    {
+                        "default": "",
+                        "tooltip": (
+                            "Path to a local folder of images. '~' expands to "
+                            "your home directory."
+                        ),
+                    },
+                ),
+                "pattern": (
+                    "STRING",
+                    {
+                        "default": _DEFAULT_FOLDER_PATTERN,
+                        "tooltip": (
+                            "Comma-separated glob pattern(s) selecting which "
+                            "files to load, e.g. '*.png,*.jpg'."
+                        ),
+                    },
+                ),
+                "max_images": (
+                    "INT",
+                    {
+                        "default": 100,
+                        "min": 1,
+                        "max": 1000,
+                        "step": 1,
+                        "tooltip": "Maximum number of images to load from the folder.",
+                    },
+                ),
+                "sort": (
+                    ["name", "modified"],
+                    {
+                        "default": "name",
+                        "tooltip": (
+                            "Order in which files are loaded: alphabetical by "
+                            "'name' or oldest-first by 'modified' time."
+                        ),
+                    },
+                ),
+            },
+        }
+
+    def load(
+        self, folder_path: str, pattern: str, max_images: int, sort: str
+    ) -> tuple[torch.Tensor, int]:
+        node_name = "FalLoadImageFolder"
+        folder = _resolve_folder(node_name, folder_path)
+        patterns = _split_csv(pattern) or _split_csv(_DEFAULT_FOLDER_PATTERN)
+        files = _sort_files(_glob_folder_files(folder, patterns), sort)[:max_images]
+        if not files:
+            raise FalApiError(
+                node_name,
+                f"No files matching '{', '.join(patterns)}' found in '{folder}'. "
+                "Adjust 'pattern' or point 'folder_path' at a folder with images.",
+            )
+        images = [self._open_image(node_name, path) for path in files]
+        batch = _images_to_batch_tensor(node_name, images, files)
+        return (batch, len(files))
+
+    @staticmethod
+    def _open_image(node_name: str, path: str) -> Image.Image:
+        """Open a local image file as RGB, normalizing failures."""
+        try:
+            with Image.open(path) as img:
+                return img.convert("RGB")
+        except Exception as exc:
+            logger.error("%s: failed to open image %s: %s", node_name, path, exc)
+            raise FalApiError(
+                node_name,
+                f"Failed to open image '{path}': {exc}. Remove or exclude the "
+                "file via 'pattern' and retry.",
+            ) from exc
+
+
+def _normalize_extensions(extensions: str) -> list[str] | None:
+    """Parse a comma-separated extension filter; empty means no filter."""
+    parts = [part.lstrip("*").lower() for part in _split_csv(extensions)]
+    normalized = [part if part.startswith(".") else f".{part}" for part in parts]
+    cleaned = [part for part in normalized if part != "."]
+    return cleaned or None
+
+
+class FalUploadFolderAsZip:
+    """Zip a local folder and upload the archive to fal.ai, returning its URL."""
+
+    RETURN_TYPES = ("STRING",)
+    RETURN_NAMES = ("zip_url",)
+    FUNCTION = "upload"
+    CATEGORY = _CATEGORY
+    DESCRIPTION = (
+        "Zip a local folder (optionally recursive / filtered by extension) and "
+        "upload the archive to fal.ai storage, returning the ZIP's URL — handy "
+        "for endpoints that take a training-data archive."
+    )
+
+    @classmethod
+    def INPUT_TYPES(cls) -> dict[str, Any]:
+        return {
+            "required": {
+                "folder_path": (
+                    "STRING",
+                    {
+                        "default": "",
+                        "tooltip": (
+                            "Path to the local folder to zip and upload. '~' "
+                            "expands to your home directory."
+                        ),
+                    },
+                ),
+                "recursive": (
+                    "BOOLEAN",
+                    {
+                        "default": False,
+                        "tooltip": "Include files from subfolders in the ZIP.",
+                    },
+                ),
+                "extensions": (
+                    "STRING",
+                    {
+                        "default": "",
+                        "tooltip": (
+                            "Comma-separated file extensions to include, e.g. "
+                            "'.png,.jpg'. Leave empty to include all files."
+                        ),
+                    },
+                ),
+            },
+        }
+
+    def upload(
+        self, folder_path: str, recursive: bool, extensions: str
+    ) -> tuple[str]:
+        node_name = "FalUploadFolderAsZip"
+        folder = _resolve_folder(node_name, folder_path)
+        archive_utils = self._load_archive_utils(node_name)
+        include_extensions = _normalize_extensions(extensions)
+        try:
+            zip_path = archive_utils.zip_folder(
+                folder, include_extensions=include_extensions, recursive=recursive
+            )
+            return (archive_utils.upload_zip(zip_path),)
+        except FalApiError:
+            raise
+        except Exception as exc:
+            logger.error("%s: failed to zip/upload %s: %s", node_name, folder, exc)
+            raise FalApiError(
+                node_name,
+                f"Failed to zip and upload folder '{folder}': {exc}",
+            ) from exc
+
+    @staticmethod
+    def _load_archive_utils(node_name: str) -> Any:
+        """Lazily import ArchiveUtils, degrading with an actionable error."""
+        try:
+            from .fal_utils import ArchiveUtils
+
+            return ArchiveUtils
+        except ImportError as exc:
+            raise FalApiError(
+                node_name,
+                "Archive utilities are unavailable in this install "
+                f"({exc}). Update/reinstall ComfyUI-fal-API so that "
+                "nodes/utils/archive.py is present.",
+            ) from exc
+
+
+NODE_CLASS_MAPPINGS = {
+    "FalLoadImageURL_fal": FalLoadImageURL,
+    "FalLoadAudioURL_fal": FalLoadAudioURL,
+    "FalLoadImageFolder_fal": FalLoadImageFolder,
+    "FalUploadFolderAsZip_fal": FalUploadFolderAsZip,
+}
+
+NODE_DISPLAY_NAME_MAPPINGS = {
+    "FalLoadImageURL_fal": "Load Image from URL (fal)",
+    "FalLoadAudioURL_fal": "Load Audio from URL (fal)",
+    "FalLoadImageFolder_fal": "Load Image Folder (fal)",
+    "FalUploadFolderAsZip_fal": "Upload Folder as ZIP URL (fal)",
+}

nodes/util_video_node.py

@@ -0,0 +1,800 @@
+"""Local video utility nodes: frame extraction, trim, concat, mux, audio extraction.
+
+These nodes run entirely locally (cv2/PyAV) — no fal.ai API calls — and are
+meant to glue video-generation workflows together (e.g. grab the last frame of
+a clip and feed it into an image-to-video node to extend the video).
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+from fractions import Fraction
+from typing import Any
+
+import numpy as np
+import torch
+
+from .fal_utils import FalApiError, MediaUtils, logger
+
+_CATEGORY = "FAL/Utils/Video"
+
+_CHUNK_SIZE = 1 << 20  # 1 MiB
+_AV_TIME_BASE = 1_000_000  # PyAV container.seek() offset units (microseconds)
+_AAC_FRAME_SIZE = 1024  # samples per AAC frame
+_TIME_EPS = 1e-6
+
+
+# ---------------------------------------------------------------------------
+# Shared helpers
+# ---------------------------------------------------------------------------
+
+
+def _safe_unlink(path: str | None) -> None:
+    """Delete a temp file, ignoring errors."""
+    if path is None:
+        return
+    try:
+        os.unlink(path)
+    except OSError:
+        pass
+
+
+def _import_cv2(node_name: str) -> Any:
+    """Import cv2 lazily with a clear error when missing."""
+    try:
+        import cv2
+
+        return cv2
+    except ImportError as exc:
+        raise FalApiError(
+            node_name, "OpenCV is required for this node — pip install opencv-python"
+        ) from exc
+
+
+def _import_av(node_name: str) -> Any:
+    """Import PyAV lazily with a clear error when missing."""
+    try:
+        import av
+
+        return av
+    except ImportError as exc:
+        raise FalApiError(node_name, "PyAV is required for this node — pip install av") from exc
+
+
+def _resolve_video_from_file() -> type | None:
+    """Locate ComfyUI's VideoFromFile class across API layouts."""
+    try:
+        from comfy_api.input_impl import VideoFromFile
+
+        return VideoFromFile
+    except ImportError:
+        pass
+    try:
+        from comfy_api.latest import input_impl
+
+        return getattr(input_impl, "VideoFromFile", None)
+    except ImportError:
+        return None
+
+
+def _wrap_local_video(path: str, node_name: str) -> Any:
+    """Wrap a local video file as a ComfyUI VIDEO object."""
+    video_cls = _resolve_video_from_file()
+    if video_cls is None:
+        raise FalApiError(
+            node_name,
+            "comfy_api VideoFromFile is unavailable; update ComfyUI to a version "
+            "that provides comfy_api to use VIDEO outputs.",
+        )
+    return video_cls(path)
+
+
+def _new_temp_path(suffix: str) -> str:
+    """Create an empty named temp file and return its path."""
+    with tempfile.NamedTemporaryFile(suffix=suffix, prefix="fal_util_video_", delete=False) as temp_file:
+        return temp_file.name
+
+
+def _spool_stream_to_temp(source: Any, node_name: str) -> str:
+    """Write a readable stream to a temp .mp4 file and return its path."""
+    temp_path: str | None = None
+    try:
+        with tempfile.NamedTemporaryFile(suffix=".mp4", prefix="fal_util_video_", delete=False) as temp_file:
+            temp_path = temp_file.name
+            while True:
+                chunk = source.read(_CHUNK_SIZE)
+                if not chunk:
+                    break
+                temp_file.write(chunk)
+        return temp_path
+    except Exception as exc:
+        _safe_unlink(temp_path)
+        raise FalApiError(node_name, f"Failed to read video stream: {exc}") from exc
+
+
+def _video_input_to_path(video: Any, node_name: str) -> tuple[str, bool]:
+    """Resolve a VIDEO input (or path/URL string) to a local file path.
+
+    Returns (path, cleanup_needed). cleanup_needed is True when the path is a
+    temp file created here that the caller must delete when done.
+    """
+    if video is None:
+        raise FalApiError(node_name, "No video input provided")
+
+    source = video.get_stream_source() if hasattr(video, "get_stream_source") else video
+
+    if isinstance(source, str):
+        if source.startswith(("http://", "https://")):
+            return MediaUtils.download_url_to_temp(source, ".mp4"), True
+        if os.path.isfile(source):
+            return source, False
+        raise FalApiError(node_name, f"Video path does not exist: {source}")
+
+    if hasattr(source, "read"):
+        return _spool_stream_to_temp(source, node_name), True
+
+    raise FalApiError(
+        node_name,
+        f"Unsupported video input of type {type(video).__name__}; expected a "
+        "VIDEO object, a local file path, or an http(s) URL string.",
+    )
+
+
+def _add_stream_from_template(output: Any, template: Any) -> Any:
+    """Add an output stream copying the template's codec parameters."""
+    if hasattr(output, "add_stream_from_template"):
+        return output.add_stream_from_template(template)
+    return output.add_stream(template=template)
+
+
+def _stream_duration_seconds(container: Any, stream: Any) -> float:
+    """Best-effort duration (seconds) of a stream, falling back to container."""
+    if stream.duration is not None and stream.time_base is not None:
+        return float(stream.duration * stream.time_base)
+    if container.duration is not None:
+        return float(container.duration) / _AV_TIME_BASE
+    return 0.0
+
+
+def _encode_audio_array(
+    av: Any, output: Any, stream: Any, layout: str, sample_rate: int, samples: np.ndarray, start_index: int
+) -> int:
+    """Encode a planar float32 (C, T) array as AAC frames; returns next sample index."""
+    total = samples.shape[1]
+    for offset in range(0, total, _AAC_FRAME_SIZE):
+        chunk = np.ascontiguousarray(samples[:, offset : offset + _AAC_FRAME_SIZE])
+        frame = av.AudioFrame.from_ndarray(chunk, format="fltp", layout=layout)
+        frame.sample_rate = sample_rate
+        frame.pts = start_index + offset
+        for packet in stream.encode(frame):
+            output.mux(packet)
+    return start_index + total
+
+
+def _pad_or_truncate(samples: np.ndarray, needed: int) -> np.ndarray:
+    """Pad a planar (C, T) array with silence, or truncate, to exactly `needed` samples."""
+    if samples.shape[1] >= needed:
+        return samples[:, :needed]
+    pad = np.zeros((samples.shape[0], needed - samples.shape[1]), dtype=np.float32)
+    return np.concatenate([samples, pad], axis=1)
+
+
+def _normalize_audio_frame(array: np.ndarray, channels: int) -> np.ndarray:
+    """Normalize a PyAV audio frame array to float32 with shape (C, N)."""
+    if np.issubdtype(array.dtype, np.integer):
+        info = np.iinfo(array.dtype)
+        scale = float(max(abs(info.min), info.max))
+        array = array.astype(np.float32) / scale
+    else:
+        array = array.astype(np.float32)
+
+    if array.ndim == 1:
+        array = array[np.newaxis, :]
+    if array.shape[0] == 1 and channels > 1:
+        # Packed/interleaved format: (1, N * C) -> (C, N)
+        array = array.reshape(-1, channels).T
+    return array
+
+
+def _waveform_to_planar(audio: Any, node_name: str) -> tuple[np.ndarray, int]:
+    """Convert a ComfyUI AUDIO dict to (planar float32 (C, T) with C in {1, 2}, sample_rate)."""
+    try:
+        waveform = audio["waveform"]
+        sample_rate = int(audio["sample_rate"])
+    except (KeyError, TypeError) as exc:
+        raise FalApiError(
+            node_name, "Expected an AUDIO dict with 'waveform' and 'sample_rate'"
+        ) from exc
+    if not isinstance(waveform, torch.Tensor):
+        raise FalApiError(node_name, "AUDIO 'waveform' must be a torch tensor")
+
+    tensor = waveform.detach().cpu().to(torch.float32)
+    if tensor.ndim == 3:
+        tensor = tensor[0]  # (B, C, T) -> (C, T)
+    if tensor.ndim == 1:
+        tensor = tensor.unsqueeze(0)
+    if tensor.ndim != 2:
+        raise FalApiError(node_name, f"AUDIO waveform has unsupported shape {tuple(waveform.shape)}")
+
+    array = tensor.clamp(-1.0, 1.0).numpy()
+    if array.shape[0] > 2:
+        logger.warning("%s: waveform has %d channels; keeping the first two", node_name, array.shape[0])
+        array = array[:2]
+    return np.ascontiguousarray(array), sample_rate
+
+
+# ---------------------------------------------------------------------------
+# FalExtractFrames
+# ---------------------------------------------------------------------------
+
+
+def _bgr_frames_to_tensor(frames: list[np.ndarray], cv2: Any) -> torch.Tensor:
+    """Convert BGR uint8 frames to a (N, H, W, 3) float32 RGB tensor in 0-1."""
+    rgb = [cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) for frame in frames]
+    stacked = np.stack(rgb).astype(np.float32) / 255.0
+    return torch.from_numpy(stacked)
+
+
+def _frame_via_seek(cap: Any, cv2: Any, index: int) -> np.ndarray | None:
+    """Seek to a frame index and read it; returns None when the seek misbehaves."""
+    cap.set(cv2.CAP_PROP_POS_FRAMES, float(index))
+    ok, frame = cap.read()
+    return frame if ok and frame is not None else None
+
+
+def _scan_frames(cap: Any, cv2: Any, stop_after: int | None = None) -> tuple[np.ndarray | None, int]:
+    """Sequentially decode from frame 0; returns (last frame seen, frames read).
+
+    Stops after reading `stop_after + 1` frames when `stop_after` is given.
+    """
+    cap.set(cv2.CAP_PROP_POS_FRAMES, 0.0)
+    last: np.ndarray | None = None
+    count = 0
+    while True:
+        ok, frame = cap.read()
+        if not ok or frame is None:
+            break
+        last = frame
+        count += 1
+        if stop_after is not None and count > stop_after:
+            break
+    return last, count
+
+
+class FalExtractFrames:
+    """Extract frames from a video as IMAGE outputs (local decode, no API call)."""
+
+    RETURN_TYPES = ("IMAGE", "INT")
+    RETURN_NAMES = ("frames", "frame_count")
+    FUNCTION = "extract_frames"
+    CATEGORY = _CATEGORY
+    DESCRIPTION = (
+        "Decode a video locally and extract frames. Mode 'last' grabs the final "
+        "frame — feed it into an image-to-video node to extend/continue a video."
+    )
+
+    @classmethod
+    def INPUT_TYPES(cls) -> dict[str, Any]:
+        return {
+            "required": {
+                "video": ("VIDEO", {"tooltip": "Video to decode. Tip: use mode 'last' to grab the final frame and feed it into an image-to-video node to extend the video."}),
+                "mode": (["first", "last", "nth", "every_nth"], {"default": "last", "tooltip": "first/last: single frame. nth: the n-th frame (1-based). every_nth: every n-th frame as a batch, capped at max_frames."}),
+                "n": ("INT", {"default": 1, "min": 1, "max": 1_000_000, "tooltip": "Frame index (1-based) for mode 'nth'; step size for 'every_nth'. Ignored otherwise."}),
+                "max_frames": ("INT", {"default": 64, "min": 1, "max": 1024, "tooltip": "Maximum number of frames returned by mode 'every_nth'; ignored for other modes."}),
+            },
+        }
+
+    def extract_frames(self, video: Any, mode: str, n: int, max_frames: int) -> tuple[torch.Tensor, int]:
+        node_name = "FalExtractFrames"
+        cv2 = _import_cv2(node_name)
+        path, cleanup = _video_input_to_path(video, node_name)
+        cap = None
+        try:
+            cap = cv2.VideoCapture(path)
+            if not cap.isOpened():
+                raise FalApiError(node_name, f"OpenCV could not open video: {path}")
+            reported = int(cap.get(cv2.CAP_PROP_FRAME_COUNT) or 0)
+
+            if mode == "every_nth":
+                frames, frame_count = self._extract_every_nth(cap, cv2, n, max_frames, reported)
+            else:
+                frame, frame_count = self._extract_single(cap, cv2, mode, n, reported, node_name)
+                frames = [frame]
+
+            if not frames or frames[0] is None:
+                raise FalApiError(node_name, f"No frames could be decoded from {path}")
+            return (_bgr_frames_to_tensor(frames, cv2), frame_count)
+        except FalApiError:
+            raise
+        except Exception as exc:
+            logger.error("FalExtractFrames failed: %s", exc)
+            raise FalApiError(node_name, f"Frame extraction failed: {exc}") from exc
+        finally:
+            if cap is not None:
+                cap.release()
+            if cleanup:
+                _safe_unlink(path)
+
+    @staticmethod
+    def _extract_every_nth(
+        cap: Any, cv2: Any, step: int, max_frames: int, reported: int
+    ) -> tuple[list[np.ndarray], int]:
+        """Sequentially collect every `step`-th frame, capped at max_frames."""
+        cap.set(cv2.CAP_PROP_POS_FRAMES, 0.0)
+        frames: list[np.ndarray] = []
+        count = 0
+        reached_eof = False
+        while True:
+            ok, frame = cap.read()
+            if not ok or frame is None:
+                reached_eof = True
+                break
+            if count % step == 0 and len(frames) < max_frames:
+                frames.append(frame)
+            count += 1
+            if len(frames) >= max_frames and reported > 0:
+                break  # early stop: reported count stands in for the true total
+        frame_count = count if reached_eof or reported <= 0 else reported
+        return frames, frame_count
+
+    @staticmethod
+    def _extract_single(
+        cap: Any, cv2: Any, mode: str, n: int, reported: int, node_name: str
+    ) -> tuple[np.ndarray | None, int]:
+        """Extract a single frame for modes first/last/nth."""
+        if mode == "first":
+            target = 0
+        elif mode == "nth":
+            target = n - 1
+            if reported > 0 and target >= reported:
+                logger.warning("%s: frame %d beyond end (%d frames); using last frame", node_name, n, reported)
+                target = reported - 1
+        elif mode == "last":
+            target = max(reported - 1, 0)
+        else:
+            raise FalApiError(node_name, f"Unknown mode: {mode}")
+
+        frame: np.ndarray | None = None
+        if reported > 0:
+            # Fast path: direct seek (some codecs mis-seek; fall back below).
+            frame = _frame_via_seek(cap, cv2, target)
+        if frame is not None:
+            return frame, reported
+
+        # Sequential fallback: decode from the start.
+        if mode == "last":
+            frame, count = _scan_frames(cap, cv2)
+            return frame, count
+        frame, read = _scan_frames(cap, cv2, stop_after=target)
+        if read > target:
+            # Reached the target; total count comes from metadata or a full scan.
+            frame_count = reported if reported > 0 else _scan_frames(cap, cv2)[1]
+            return frame, frame_count
+        # Hit EOF early: `frame` is the last decodable frame, `read` the true count.
+        return frame, read
+
+
+# ---------------------------------------------------------------------------
+# FalTrimVideo
+# ---------------------------------------------------------------------------
+
+
+def _find_seek_start(av: Any, container: Any, anchor: Any, start: float) -> float:
+    """Seek near `start` and return the timestamp of the first packet (keyframe snap)."""
+    if start <= 0:
+        return 0.0
+    container.seek(int(start * _AV_TIME_BASE), backward=True, any_frame=False)
+    actual_start = start
+    for packet in container.demux(anchor):
+        if packet.pts is None:
+            continue
+        actual_start = float(packet.pts * packet.time_base)
+        break
+    container.seek(int(start * _AV_TIME_BASE), backward=True, any_frame=False)
+    return actual_start
+
+
+def _remux_trim(av: Any, in_path: str, out_path: str, start: float, end: float | None, node_name: str) -> None:
+    """Copy packets between timestamps into a new mp4 without re-encoding."""
+    with av.open(in_path) as container, av.open(out_path, mode="w") as output:
+        video_in = container.streams.video[0] if container.streams.video else None
+        audio_in = container.streams.audio[0] if container.streams.audio else None
+        selected = [stream for stream in (video_in, audio_in) if stream is not None]
+        if not selected:
+            raise FalApiError(node_name, "Input has no video or audio streams")
+
+        out_streams = {stream.index: _add_stream_from_template(output, stream) for stream in selected}
+        anchor = video_in if video_in is not None else audio_in
+        actual_start = _find_seek_start(av, container, anchor, start)
+        if end is not None and end <= actual_start + _TIME_EPS:
+            raise FalApiError(
+                node_name,
+                f"Trim range is empty: start snapped to keyframe at {actual_start:.3f}s, end is {end:.3f}s",
+            )
+
+        offsets: dict[int, int] = {}
+        done = dict.fromkeys(out_streams, False)
+        kept = 0
+        for packet in container.demux(selected):
+            if packet.pts is None:
+                continue
+            index = packet.stream.index
+            if done.get(index, True):
+                continue
+            time = float(packet.pts * packet.time_base)
+            if time < actual_start - _TIME_EPS:
+                continue
+            if end is not None and time >= end - _TIME_EPS:
+                done[index] = True
+                if all(done.values()):
+                    break
+                continue
+            if index not in offsets:
+                offsets[index] = packet.dts if packet.dts is not None else packet.pts
+            offset = offsets[index]
+            packet.pts -= offset
+            if packet.dts is not None:
+                packet.dts -= offset
+            packet.stream = out_streams[index]
+            output.mux(packet)
+            kept += 1
+
+        if kept == 0:
+            raise FalApiError(node_name, f"Trim produced no packets (start {start:.3f}s may be past the end)")
+
+
+class FalTrimVideo:
+    """Trim a video to [start, end] seconds by remuxing (no re-encode)."""
+
+    RETURN_TYPES = ("VIDEO", "STRING")
+    RETURN_NAMES = ("video", "path")
+    FUNCTION = "trim_video"
+    CATEGORY = _CATEGORY
+    DESCRIPTION = (
+        "Trim a video without re-encoding by copying packets between timestamps. "
+        "Fast and lossless, but the start cut snaps to the nearest earlier keyframe."
+    )
+
+    @classmethod
+    def INPUT_TYPES(cls) -> dict[str, Any]:
+        return {
+            "required": {
+                "video": ("VIDEO", {"tooltip": "Video to trim (video + audio tracks are kept)."}),
+                "start_seconds": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 100_000.0, "step": 0.1, "tooltip": "Trim start in seconds. Cuts snap to the nearest earlier keyframe (no re-encode)."}),
+                "end_seconds": ("FLOAT", {"default": 5.0, "min": 0.0, "max": 100_000.0, "step": 0.1, "tooltip": "Trim end in seconds; 0 = to the end of the video."}),
+            },
+        }
+
+    def trim_video(self, video: Any, start_seconds: float, end_seconds: float) -> tuple[Any, str]:
+        node_name = "FalTrimVideo"
+        av = _import_av(node_name)
+        end = end_seconds if end_seconds > 0 else None
+        if end is not None and end <= start_seconds:
+            raise FalApiError(node_name, f"end_seconds ({end:.3f}) must be greater than start_seconds ({start_seconds:.3f})")
+        path, cleanup = _video_input_to_path(video, node_name)
+        out_path = _new_temp_path(".mp4")
+        try:
+            _remux_trim(av, path, out_path, start_seconds, end, node_name)
+            # NOTE: out_path is deliberately kept — VideoFromFile reads it lazily.
+            return (_wrap_local_video(out_path, node_name), out_path)
+        except FalApiError:
+            _safe_unlink(out_path)
+            raise
+        except Exception as exc:
+            _safe_unlink(out_path)
+            logger.error("FalTrimVideo failed: %s", exc)
+            raise FalApiError(node_name, f"Trim failed: {exc}") from exc
+        finally:
+            if cleanup:
+                _safe_unlink(path)
+
+
+# ---------------------------------------------------------------------------
+# FalConcatVideos
+# ---------------------------------------------------------------------------
+
+
+def _probe_video(av: Any, path: str, node_name: str) -> dict[str, Any]:
+    """Probe a clip for resolution, fps, audio presence, and audio rate."""
+    with av.open(path) as container:
+        if not container.streams.video:
+            raise FalApiError(node_name, f"Input has no video stream: {path}")
+        stream = container.streams.video[0]
+        fps = stream.average_rate or stream.guessed_rate
+        if not fps or fps <= 0:
+            fps = Fraction(30, 1)
+        audio_rate = None
+        if container.streams.audio:
+            audio_rate = int(container.streams.audio[0].rate or 44100)
+        return {
+            "width": max(2, stream.width - stream.width % 2),
+            "height": max(2, stream.height - stream.height % 2),
+            "fps": Fraction(fps),
+            "audio_rate": audio_rate,
+        }
+
+
+def _encode_video_frame(output: Any, stream: Any, frame: Any, width: int, height: int, time_base: Fraction, index: int) -> None:
+    """Scale/convert one decoded frame and encode it at the given frame index."""
+    scaled = frame.reformat(width=width, height=height, format="yuv420p")
+    scaled.pts = index
+    scaled.time_base = time_base
+    for packet in stream.encode(scaled):
+        output.mux(packet)
+
+
+def _append_clip_video(
+    av: Any, output: Any, stream: Any, path: str, width: int, height: int, fps: Fraction, start_index: int, node_name: str
+) -> int:
+    """Decode a clip, resample to target fps/size, encode; returns frames emitted."""
+    time_base = Fraction(1, 1) / fps
+    step = 1.0 / float(fps)
+    emitted = 0
+    with av.open(path) as container:
+        next_time = 0.0
+        last = None
+        for frame in container.decode(container.streams.video[0]):
+            time = frame.time if frame.time is not None else next_time
+            while last is not None and time > next_time + _TIME_EPS:
+                _encode_video_frame(output, stream, last, width, height, time_base, start_index + emitted)
+                emitted += 1
+                next_time += step
+            last = frame
+        if last is not None:
+            _encode_video_frame(output, stream, last, width, height, time_base, start_index + emitted)
+            emitted += 1
+    if emitted == 0:
+        raise FalApiError(node_name, f"No video frames decoded from {path}")
+    return emitted
+
+
+def _clip_audio_samples(av: Any, path: str, rate: int, needed: int) -> np.ndarray:
+    """Decode+resample a clip's audio to stereo float32 (2, needed); silence when absent."""
+    with av.open(path) as container:
+        if not container.streams.audio:
+            return np.zeros((2, needed), dtype=np.float32)
+        resampler = av.AudioResampler(format="fltp", layout="stereo", rate=rate)
+        chunks: list[np.ndarray] = []
+        for frame in container.decode(container.streams.audio[0]):
+            frame.pts = None  # let the resampler track timestamps itself
+            chunks.extend(out.to_ndarray() for out in resampler.resample(frame))
+        chunks.extend(out.to_ndarray() for out in resampler.resample(None))
+    if not chunks:
+        return np.zeros((2, needed), dtype=np.float32)
+    samples = np.concatenate(chunks, axis=1).astype(np.float32)
+    return _pad_or_truncate(samples, needed)
+
+
+class FalConcatVideos:
+    """Concatenate 2-4 videos by re-encoding to the first clip's resolution and fps."""
+
+    RETURN_TYPES = ("VIDEO", "STRING")
+    RETURN_NAMES = ("video", "path")
+    FUNCTION = "concat_videos"
+    CATEGORY = _CATEGORY
+    DESCRIPTION = (
+        "Concatenate up to 4 videos. All clips are re-encoded (h264 crf 18) and scaled to "
+        "video_1's resolution and fps, so mismatched codecs/sizes are fine. Audio: the output "
+        "gets a stereo AAC track when any input has audio; inputs without audio contribute silence."
+    )
+
+    @classmethod
+    def INPUT_TYPES(cls) -> dict[str, Any]:
+        return {
+            "required": {
+                "video_1": ("VIDEO", {"tooltip": "First clip; its resolution and fps define the output format."}),
+                "video_2": ("VIDEO", {"tooltip": "Second clip, scaled to match video_1."}),
+            },
+            "optional": {
+                "video_3": ("VIDEO", {"tooltip": "Optional third clip."}),
+                "video_4": ("VIDEO", {"tooltip": "Optional fourth clip."}),
+            },
+        }
+
+    def concat_videos(
+        self, video_1: Any, video_2: Any, video_3: Any = None, video_4: Any = None
+    ) -> tuple[Any, str]:
+        node_name = "FalConcatVideos"
+        av = _import_av(node_name)
+
+        inputs = [video for video in (video_1, video_2, video_3, video_4) if video is not None]
+        resolved: list[tuple[str, bool]] = []
+        out_path = _new_temp_path(".mp4")
+        try:
+            resolved = [_video_input_to_path(video, node_name) for video in inputs]
+            paths = [path for path, _ in resolved]
+            probes = [_probe_video(av, path, node_name) for path in paths]
+
+            target = probes[0]
+            width, height, fps = target["width"], target["height"], target["fps"]
+            audio_rates = [probe["audio_rate"] for probe in probes if probe["audio_rate"]]
+            audio_rate = audio_rates[0] if audio_rates else None
+
+            with av.open(out_path, mode="w") as output:
+                video_out = output.add_stream("libx264", rate=fps, options={"crf": "18", "preset": "veryfast"})
+                video_out.width = width
+                video_out.height = height
+                video_out.pix_fmt = "yuv420p"
+                audio_out = None
+                if audio_rate is not None:
+                    audio_out = output.add_stream("aac", rate=audio_rate)
+                    audio_out.layout = "stereo"
+
+                frame_index = 0
+                sample_index = 0
+                for path in paths:
+                    emitted = _append_clip_video(av, output, video_out, path, width, height, fps, frame_index, node_name)
+                    frame_index += emitted
+                    if audio_out is not None:
+                        needed = round(emitted / float(fps) * audio_rate)
+                        samples = _clip_audio_samples(av, path, audio_rate, needed)
+                        sample_index = _encode_audio_array(av, output, audio_out, "stereo", audio_rate, samples, sample_index)
+
+                for packet in video_out.encode(None):
+                    output.mux(packet)
+                if audio_out is not None:
+                    for packet in audio_out.encode(None):
+                        output.mux(packet)
+
+            # NOTE: out_path is deliberately kept — VideoFromFile reads it lazily.
+            return (_wrap_local_video(out_path, node_name), out_path)
+        except FalApiError:
+            _safe_unlink(out_path)
+            raise
+        except Exception as exc:
+            _safe_unlink(out_path)
+            logger.error("FalConcatVideos failed: %s", exc)
+            raise FalApiError(node_name, f"Concat failed: {exc}") from exc
+        finally:
+            for path, cleanup in resolved:
+                if cleanup:
+                    _safe_unlink(path)
+
+
+# ---------------------------------------------------------------------------
+# FalMuxAudioVideo
+# ---------------------------------------------------------------------------
+
+
+class FalMuxAudioVideo:
+    """Mux an AUDIO waveform onto a video, replacing any existing audio track."""
+
+    RETURN_TYPES = ("VIDEO", "STRING")
+    RETURN_NAMES = ("video", "path")
+    FUNCTION = "mux_audio_video"
+    CATEGORY = _CATEGORY
+    DESCRIPTION = (
+        "Attach an AUDIO input to a video as an AAC track, replacing any existing audio. "
+        "Video packets are copied without re-encoding."
+    )
+
+    @classmethod
+    def INPUT_TYPES(cls) -> dict[str, Any]:
+        return {
+            "required": {
+                "video": ("VIDEO", {"tooltip": "Video track; copied without re-encoding. Existing audio is replaced."}),
+                "audio": ("AUDIO", {"tooltip": "Audio to attach, encoded as AAC at its own sample rate."}),
+                "duration_policy": (["video", "shortest"], {"default": "video", "tooltip": "video: keep full video; audio is padded with silence or truncated to match. shortest: cut the output at whichever track ends first."}),
+            },
+        }
+
+    def mux_audio_video(self, video: Any, audio: Any, duration_policy: str) -> tuple[Any, str]:
+        node_name = "FalMuxAudioVideo"
+        av = _import_av(node_name)
+        samples, sample_rate = _waveform_to_planar(audio, node_name)
+        layout = "mono" if samples.shape[0] == 1 else "stereo"
+        audio_duration = samples.shape[1] / float(sample_rate)
+
+        path, cleanup = _video_input_to_path(video, node_name)
+        out_path = _new_temp_path(".mp4")
+        try:
+            with av.open(path) as container:
+                if not container.streams.video:
+                    raise FalApiError(node_name, "Input has no video stream")
+                video_in = container.streams.video[0]
+                video_duration = _stream_duration_seconds(container, video_in)
+                if video_duration <= 0:
+                    raise FalApiError(node_name, "Could not determine the video duration")
+                target = video_duration if duration_policy == "video" else min(video_duration, audio_duration)
+
+                with av.open(out_path, mode="w") as output:
+                    video_out = _add_stream_from_template(output, video_in)
+                    audio_out = output.add_stream("aac", rate=sample_rate)
+                    audio_out.layout = layout
+
+                    for packet in container.demux(video_in):
+                        if packet.dts is None:
+                            continue
+                        if duration_policy == "shortest" and float(packet.dts * packet.time_base) >= target - _TIME_EPS:
+                            break
+                        packet.stream = video_out
+                        output.mux(packet)
+
+                    needed = round(target * sample_rate)
+                    _encode_audio_array(
+                        av, output, audio_out, layout, sample_rate, _pad_or_truncate(samples, needed), 0
+                    )
+                    for packet in audio_out.encode(None):
+                        output.mux(packet)
+
+            # NOTE: out_path is deliberately kept — VideoFromFile reads it lazily.
+            return (_wrap_local_video(out_path, node_name), out_path)
+        except FalApiError:
+            _safe_unlink(out_path)
+            raise
+        except Exception as exc:
+            _safe_unlink(out_path)
+            logger.error("FalMuxAudioVideo failed: %s", exc)
+            raise FalApiError(node_name, f"Mux failed: {exc}") from exc
+        finally:
+            if cleanup:
+                _safe_unlink(path)
+
+
+# ---------------------------------------------------------------------------
+# FalVideoToAudio
+# ---------------------------------------------------------------------------
+
+
+class FalVideoToAudio:
+    """Extract the audio track of a video as a ComfyUI AUDIO output."""
+
+    RETURN_TYPES = ("AUDIO",)
+    RETURN_NAMES = ("audio",)
+    FUNCTION = "video_to_audio"
+    CATEGORY = _CATEGORY
+    DESCRIPTION = "Extract a video's audio track as an AUDIO output at its original sample rate."
+
+    @classmethod
+    def INPUT_TYPES(cls) -> dict[str, Any]:
+        return {
+            "required": {
+                "video": ("VIDEO", {"tooltip": "Video whose audio track should be extracted."}),
+            },
+        }
+
+    def video_to_audio(self, video: Any) -> tuple[dict[str, Any]]:
+        node_name = "FalVideoToAudio"
+        av = _import_av(node_name)
+        path, cleanup = _video_input_to_path(video, node_name)
+        try:
+            with av.open(path) as container:
+                if not container.streams.audio:
+                    raise FalApiError(node_name, "video has no audio track")
+                stream = container.streams.audio[0]
+                sample_rate = int(stream.rate or 44100)
+                channels = int(getattr(stream, "channels", 1) or 1)
+                chunks = [
+                    _normalize_audio_frame(frame.to_ndarray(), channels)
+                    for frame in container.decode(stream)
+                ]
+            if not chunks:
+                raise FalApiError(node_name, "No audio frames could be decoded")
+            waveform = torch.from_numpy(np.concatenate(chunks, axis=1)).unsqueeze(0)
+            return ({"waveform": waveform, "sample_rate": sample_rate},)
+        except FalApiError:
+            raise
+        except Exception as exc:
+            logger.error("FalVideoToAudio failed: %s", exc)
+            raise FalApiError(node_name, f"Audio extraction failed: {exc}") from exc
+        finally:
+            if cleanup:
+                _safe_unlink(path)
+
+
+NODE_CLASS_MAPPINGS = {
+    "FalExtractFrames_fal": FalExtractFrames,
+    "FalTrimVideo_fal": FalTrimVideo,
+    "FalConcatVideos_fal": FalConcatVideos,
+    "FalMuxAudioVideo_fal": FalMuxAudioVideo,
+    "FalVideoToAudio_fal": FalVideoToAudio,
+}
+
+NODE_DISPLAY_NAME_MAPPINGS = {
+    "FalExtractFrames_fal": "Extract Frames (fal)",
+    "FalTrimVideo_fal": "Trim Video (fal)",
+    "FalConcatVideos_fal": "Concat Videos (fal)",
+    "FalMuxAudioVideo_fal": "Mux Audio + Video (fal)",
+    "FalVideoToAudio_fal": "Video → Audio (fal)",
+}

nodes/utils/__init__.py

@@ -1,6 +1,7 @@
 """Core utilities for the ComfyUI-fal-API node pack."""
 
 from .api import ApiHandler
+from .archive import ArchiveUtils
 from .billing import BillingUtils, SpendGuard
 from .config import FalConfig
 from .errors import FalApiError, extract_error_message, raise_fal_error
@@ -14,6 +15,7 @@ from .result_cache import ResultCache
 
 __all__ = [
     "ApiHandler",
+    "ArchiveUtils",
     "BillingUtils",
     "FalApiError",
     "FalConfig",

nodes/utils/api.py

@@ -180,6 +180,15 @@ def _store_result_in_cache(
         logger.debug("[%s] result cache store failed: %s", endpoint, exc)
 
 
+def _remember_result_urls(endpoint: str, request_id: str | None, result: Any) -> None:
+    """Best-effort provenance bookkeeping: map result URLs to their request."""
+    if not request_id:
+        return
+    try:
+        ResultCache().remember_urls(endpoint, request_id, result)
+    except Exception as exc:
+        logger.debug("[%s] remember_urls failed: %s", endpoint, exc)
+
 def _raise_generation_error(model_name: str, error: Exception | str) -> NoReturn:
     """Normalize an exception or error string into a raised FalApiError."""
     if isinstance(error, BaseException):
@@ -253,6 +262,7 @@ class ApiHandler:
             _record_ledger_entry(endpoint, request_id_ref[0], duration_s)
 
         _store_result_in_cache(endpoint, arguments, result, request_id_ref[0])
+        _remember_result_urls(endpoint, request_id_ref[0], result)
         return result
 
     @staticmethod
@@ -329,6 +339,7 @@ class ApiHandler:
             JobStore().mark_collected(request_id)
         except Exception as exc:
             logger.debug("[%s] job store mark_collected failed: %s", endpoint, exc)
+        _remember_result_urls(endpoint, request_id, result)
         return result
 
     @staticmethod

nodes/utils/archive.py

@@ -0,0 +1,223 @@
+"""Zip archive helpers for dataset preparation (LoRA training uploads)."""
+
+from __future__ import annotations
+
+import io
+import os
+import tempfile
+import zipfile
+from typing import Any
+
+import torch
+
+from .config import FalConfig
+from .errors import FalApiError
+from .images import ImageUtils
+from .logger import logger
+
+_MODEL_NAME = "archive"
+
+
+def _safe_unlink(path: str | None) -> None:
+    """Delete a temp file, ignoring errors."""
+    if path is None:
+        return
+    try:
+        os.unlink(path)
+    except OSError:
+        pass
+
+
+def _split_frames(images: Any) -> list[Any]:
+    """Split an IMAGE input (batch tensor, list, or single image) into frames."""
+    if images is None:
+        return []
+    if isinstance(images, torch.Tensor):
+        if images.ndim == 4:
+            return [images[i] for i in range(images.shape[0])]
+        return [images]
+    if isinstance(images, (list, tuple)):
+        return list(images)
+    return [images]
+
+
+def _normalize_extensions(extensions: Any) -> list[str] | None:
+    """Normalize an extension filter to lowercase dot-prefixed suffixes."""
+    if not extensions:
+        return None
+    normalized = []
+    for ext in extensions:
+        cleaned = str(ext).strip().lower()
+        if not cleaned:
+            continue
+        normalized.append(cleaned if cleaned.startswith(".") else f".{cleaned}")
+    return normalized or None
+
+
+def _matches_filter(file_name: str, extensions: list[str] | None) -> bool:
+    """Whether a file passes the hidden-file and extension filters."""
+    if file_name.startswith("."):
+        return False
+    if extensions is None:
+        return True
+    return os.path.splitext(file_name)[1].lower() in extensions
+
+
+def _collect_folder_files(
+    folder: str, extensions: list[str] | None, recursive: bool
+) -> list[str]:
+    """List matching files in a folder (sorted, hidden entries skipped)."""
+    if not recursive:
+        return [
+            os.path.join(folder, name)
+            for name in sorted(os.listdir(folder))
+            if os.path.isfile(os.path.join(folder, name))
+            and _matches_filter(name, extensions)
+        ]
+    matches: list[str] = []
+    for root, dirs, files in os.walk(folder):
+        dirs[:] = sorted(d for d in dirs if not d.startswith("."))
+        for name in sorted(files):
+            if _matches_filter(name, extensions):
+                matches.append(os.path.join(root, name))
+    return matches
+
+
+def _new_temp_zip_path() -> str:
+    """Reserve a temp .zip path and return it."""
+    with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as temp_zip:
+        return temp_zip.name
+
+
+class ArchiveUtils:
+    """Utility functions for building and uploading zip archives."""
+
+    @staticmethod
+    def zip_images(
+        images: Any,
+        captions: list[str] | None = None,
+        name_prefix: str = "image",
+    ) -> str:
+        """Zip an IMAGE batch as image_0.png, image_1.png, ... and return the local zip path.
+
+        ``captions`` (optional, one per image, entries may be "") also writes
+        image_0.txt, image_1.txt, ... — the standard LoRA-training caption
+        layout. The caller is responsible for uploading/deleting the zip
+        (see ``upload_zip``).
+        """
+        frames = _split_frames(images)
+        if not frames:
+            raise FalApiError(
+                _MODEL_NAME,
+                "No images provided to zip. Connect an IMAGE batch with at least one frame.",
+            )
+        if captions is not None and len(captions) != len(frames):
+            raise FalApiError(
+                _MODEL_NAME,
+                f"Caption count ({len(captions)}) does not match image count ({len(frames)}). "
+                "Provide exactly one caption per image (blank entries are allowed) or none at all.",
+            )
+        prefix = (name_prefix or "image").strip() or "image"
+
+        zip_path: str | None = None
+        try:
+            zip_path = _new_temp_zip_path()
+            with zipfile.ZipFile(zip_path, "w") as zip_file:
+                for index, frame in enumerate(frames):
+                    pil_image = ImageUtils.tensor_to_pil(frame)
+                    buffer = io.BytesIO()
+                    pil_image.save(buffer, format="PNG")
+                    zip_file.writestr(f"{prefix}_{index}.png", buffer.getvalue())
+                    if captions is not None:
+                        zip_file.writestr(f"{prefix}_{index}.txt", captions[index])
+            return zip_path
+        except FalApiError:
+            _safe_unlink(zip_path)
+            raise
+        except Exception as exc:
+            _safe_unlink(zip_path)
+            logger.error("Failed to create image zip: %s", exc)
+            raise FalApiError(
+                _MODEL_NAME, f"Failed to create image zip: {exc}"
+            ) from exc
+
+    @staticmethod
+    def zip_folder(
+        folder_path: str,
+        include_extensions: list[str] | None = None,
+        recursive: bool = False,
+    ) -> str:
+        """Zip a folder's files and return the local zip path.
+
+        ``include_extensions`` filters by suffix (e.g. [".png", ".txt"]); None
+        includes everything. Hidden files/directories are always skipped.
+        Non-recursive by default; arcnames are relative to the folder.
+        """
+        if not folder_path or not isinstance(folder_path, str) or not folder_path.strip():
+            raise FalApiError(
+                _MODEL_NAME,
+                "folder_path is empty. Provide the path to a folder of dataset files.",
+            )
+        folder = os.path.abspath(os.path.expanduser(folder_path.strip()))
+        if not os.path.isdir(folder):
+            raise FalApiError(
+                _MODEL_NAME,
+                f"Folder not found: {folder}. Provide the path to an existing directory.",
+            )
+
+        extensions = _normalize_extensions(include_extensions)
+        files = _collect_folder_files(folder, extensions, recursive)
+        if not files:
+            suffix_hint = f" matching extensions {extensions}" if extensions else ""
+            raise FalApiError(
+                _MODEL_NAME,
+                f"No files{suffix_hint} found in {folder}. "
+                "Check the folder contents, the extension filter, and the recursive flag.",
+            )
+
+        # Folder zips get uploaded to fal's CDN: log loudly what is being read
+        # and cap runaway/hostile selections ([archive] section in config.ini).
+        total_bytes = sum(os.path.getsize(f) for f in files)
+        max_files = int(FalConfig().get_setting("archive", "max_files", 5000))
+        max_mb = float(FalConfig().get_setting("archive", "max_total_mb", 2048))
+        if len(files) > max_files or total_bytes > max_mb * 1024 * 1024:
+            raise FalApiError(
+                _MODEL_NAME,
+                f"Refusing to zip {len(files)} file(s) / {total_bytes / 1048576:.1f} MiB "
+                f"from {folder} — over the [archive] limits (max_files={max_files}, "
+                f"max_total_mb={max_mb:g}). Narrow the folder/extensions or raise the "
+                "limits in config.ini.",
+            )
+        logger.info(
+            "archive: zipping %d file(s) (%.1f MiB) from %s",
+            len(files),
+            total_bytes / 1048576,
+            folder,
+        )
+
+        zip_path: str | None = None
+        try:
+            zip_path = _new_temp_zip_path()
+            with zipfile.ZipFile(zip_path, "w") as zip_file:
+                for file_path in files:
+                    zip_file.write(file_path, os.path.relpath(file_path, folder))
+            return zip_path
+        except Exception as exc:
+            _safe_unlink(zip_path)
+            logger.error("Failed to zip folder %s: %s", folder, exc)
+            raise FalApiError(
+                _MODEL_NAME, f"Failed to zip folder {folder}: {exc}"
+            ) from exc
+
+    @staticmethod
+    def upload_zip(zip_path: str) -> str:
+        """Upload a local zip to fal.ai and return its URL; the zip is always deleted."""
+        if not zip_path or not os.path.isfile(zip_path):
+            raise FalApiError(
+                _MODEL_NAME,
+                f"Zip file not found: {zip_path}. Build it with zip_images/zip_folder first.",
+            )
+        try:
+            return ImageUtils.upload_file(zip_path)
+        finally:
+            _safe_unlink(zip_path)

nodes/utils/billing.py

@@ -139,16 +139,16 @@ class BillingUtils:
         as SpendGuard.preflight do not hammer the API; ``force=True`` bypasses.
         """
         global _balance_cache
-        now = time.time()
-        if not force:
-            with _balance_lock:
-                value, fetched_at = _balance_cache
-            if fetched_at > 0 and now - fetched_at < _BALANCE_CACHE_TTL_S:
-                return value
-        value = _fetch_balance()
+        # the fetch happens under the lock so a cold-start burst of parallel
+        # callers (e.g. 8 caption workers hitting SpendGuard at once) collapses
+        # into a single API call instead of hammering /account/billing
         with _balance_lock:
+            value, fetched_at = _balance_cache
+            if not force and fetched_at > 0 and time.time() - fetched_at < _BALANCE_CACHE_TTL_S:
+                return value
+            value = _fetch_balance()
             _balance_cache = [value, time.time()]
-        return value
+            return value
 
     @staticmethod
     def get_recent_usage(limit: int = 50) -> list[dict[str, Any]] | None:

nodes/utils/result_cache.py

@@ -50,6 +50,14 @@ _SCHEMA = (
         created REAL
     )
     """,
+    """
+    CREATE TABLE IF NOT EXISTS request_urls (
+        url TEXT PRIMARY KEY,
+        endpoint TEXT,
+        request_id TEXT,
+        created REAL
+    )
+    """,
 )
 
 
@@ -301,17 +309,65 @@ class ResultCache:
             "misses": self._misses,
         }
 
+    def remember_urls(self, endpoint: str, request_id: str, result: Any) -> None:
+        """Record every media URL in a result → (endpoint, request_id).
+
+        Called on every successful fetch (live, async collect, recovery) so
+        provenance lookups work regardless of which path produced the result.
+        Best-effort: never raises.
+        """
+        try:
+            if not request_id or not isinstance(result, dict):
+                return
+            urls: list[str] = []
+
+            def dig(value: Any) -> None:
+                if isinstance(value, dict):
+                    candidate = value.get("url")
+                    if isinstance(candidate, str) and candidate.startswith("http"):
+                        urls.append(candidate)
+                    for child in value.values():
+                        dig(child)
+                elif isinstance(value, list):
+                    for child in value:
+                        dig(child)
+
+            dig(result)
+            if not urls:
+                return
+            now = time.time()
+            with self._lock:
+                conn = self._connection()
+                if conn is None:
+                    return
+                conn.executemany(
+                    "INSERT OR REPLACE INTO request_urls VALUES (?, ?, ?, ?)",
+                    [(u, endpoint, request_id, now) for u in urls[:64]],
+                )
+                conn.commit()
+        except Exception as exc:
+            logger.debug("result cache remember_urls failed: %s", exc)
+
     def find_request_by_url(self, url: str) -> dict[str, Any] | None:
         """Find the origin of a result URL: {"endpoint_id", "request_id"} or None.
 
-        Scans cached results (most recently used first) for one whose JSON
-        contains ``url`` as an exact substring. Only rows that recorded a
-        request_id qualify. Best-effort: any failure is a miss.
+        Checks the explicit request_urls table first (covers async collect and
+        recovery paths), then falls back to scanning cached results for the
+        URL as a substring. Best-effort: any failure is a miss.
         """
         try:
             target = (url or "").strip()
             if not target:
                 return None
+            with self._lock:
+                conn = self._connection()
+                if conn is not None:
+                    row = conn.execute(
+                        "SELECT endpoint, request_id FROM request_urls WHERE url = ?",
+                        (target,),
+                    ).fetchone()
+                    if row is not None:
+                        return {"endpoint_id": row[0], "request_id": row[1]}
             # LIKE treats %, _ (and our escape char) specially — escape them
             # so URLs containing percent-encoding still match literally.
             escaped = (

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "fal-api"
 description = "Custom nodes for using fal API with auto-generated full-catalog coverage of fal.ai models. Video generation with Kling, Runway, Luma. Image generation with Flux. LLMs and VLMs OpenAI, Claude, Llama and Gemini."
-version = "2.3.0"
+version = "2.4.0"
 license = {file = "LICENSE"}
 requires-python = ">=3.9"
 dependencies = [
@@ -11,6 +11,7 @@ dependencies = [
     "numpy",
     "pillow",
     "requests",
+    "av",
 ]
 
 [project.urls]

requirements.txt

@@ -4,3 +4,4 @@ opencv-python
 numpy
 pillow
 requests
+av

tests/test_provenance.py

@@ -70,3 +70,16 @@ def test_png_chunk_roundtrip(platform, tmp_path):
     endpoint, request_id, _ = node.read(file_path=str(png))
     assert endpoint == "fal-ai/flux-2"
     assert request_id == "req-png"
+
+
+def test_remember_urls_covers_async_results(platform):
+    """Provenance must work for Submit→Collect results, not just cached calls."""
+    cache_mod = importlib.import_module(f"{PKG}.nodes.utils.result_cache")
+    cache = cache_mod.ResultCache()
+    cache.clear()
+    url = "https://v3.fal.media/files/x/collected_output.jpg"
+    # no cache.put() — this simulates the async-collect path
+    cache.remember_urls("fal-ai/veo3", "req-async", {"images": [{"url": url}]})
+    hit = cache.find_request_by_url(url)
+    assert hit == {"endpoint_id": "fal-ai/veo3", "request_id": "req-async"}
+    cache.clear()

tests/test_util_nodes.py

@@ -0,0 +1,120 @@
+"""Tests for the FAL/Utils node layer (dataset, image, data, video basics)."""
+
+from __future__ import annotations
+
+import importlib
+import json
+import zipfile
+
+import pytest
+import torch
+from conftest import PKG, _load_package
+
+
+@pytest.fixture(scope="session")
+def archive_mod():
+    _load_package()
+    return importlib.import_module(f"{PKG}.nodes.utils.archive")
+
+
+@pytest.fixture(scope="session")
+def image_nodes(pack):
+    return pack.NODE_CLASS_MAPPINGS
+
+
+def test_zip_images_with_captions(archive_mod, tmp_path):
+    images = torch.rand(3, 8, 8, 3)
+    zip_path = archive_mod.ArchiveUtils.zip_images(images, captions=["a", "", "c"])
+    try:
+        with zipfile.ZipFile(zip_path) as zf:
+            names = sorted(zf.namelist())
+            assert "image_0.png" in names and "image_2.txt" in names
+            assert zf.read("image_0.txt").decode() == "a"
+    finally:
+        import os
+
+        os.unlink(zip_path)
+
+
+def test_zip_images_caption_mismatch_raises(archive_mod, errors_mod):
+    with pytest.raises(errors_mod.FalApiError):
+        archive_mod.ArchiveUtils.zip_images(torch.rand(2, 8, 8, 3), captions=["only one"])
+
+
+def test_json_extract(pack):
+    cls = pack.NODE_CLASS_MAPPINGS["FalJSONExtract_fal"]
+    node = cls()
+    fn = getattr(node, cls.FUNCTION)
+    payload = json.dumps({"video": {"url": "https://x/v.mp4"}, "images": [{"url": "https://x/i.png"}], "seed": 42})
+    assert fn(json_text=payload, path="video.url", default="")[0] == "https://x/v.mp4"
+    assert fn(json_text=payload, path="images[0].url", default="")[0] == "https://x/i.png"
+    assert fn(json_text=payload, path="seed", default="")[1] == 42.0
+    assert fn(json_text=payload, path="missing.path", default="fallback")[0] == "fallback"
+
+
+def test_prompt_lines_wraps(pack):
+    cls = pack.NODE_CLASS_MAPPINGS["FalPromptLines_fal"]
+    node = cls()
+    fn = getattr(node, cls.FUNCTION)
+    text = "one\ntwo\nthree"
+    assert fn(text=text, index=0, skip_blank=True)[0] == "one"
+    assert fn(text=text, index=4, skip_blank=True)[0] == "two"  # wraps modulo 3
+
+
+def test_resize_to_preset_dims(pack):
+    cls = pack.NODE_CLASS_MAPPINGS["FalResizeToPreset_fal"]
+    node = cls()
+    fn = getattr(node, cls.FUNCTION)
+    image = torch.rand(1, 300, 500, 3)
+    out, width, height = fn(image=image, preset="landscape_16_9", width=1024, height=1024, mode="cover_crop")
+    assert (width, height) == (1024, 576)
+    assert tuple(out.shape) == (1, 576, 1024, 3)
+
+
+def test_base64_round_trip(pack):
+    cm = pack.NODE_CLASS_MAPPINGS
+    enc_cls, dec_cls = cm["FalImageToBase64_fal"], cm["FalBase64ToImage_fal"]
+    image = torch.rand(1, 16, 16, 3)
+    encoded = getattr(enc_cls(), enc_cls.FUNCTION)(image=image, format="png", data_uri=True)[0]
+    decoded = getattr(dec_cls(), dec_cls.FUNCTION)(data=encoded)[0]
+    assert tuple(decoded.shape) == (1, 16, 16, 3)
+    assert torch.allclose(image, decoded, atol=2 / 255)
+
+
+def test_image_grid_shape(pack):
+    cls = pack.NODE_CLASS_MAPPINGS["FalImageGrid_fal"]
+    node = cls()
+    fn = getattr(node, cls.FUNCTION)
+    out = fn(images=torch.rand(4, 32, 32, 3), labels="a\nb\nc\nd", columns=2, cell_padding=4, label_height=16)[0]
+    assert out.ndim == 4 and out.shape[0] == 1 and out.shape[3] == 3
+
+
+def test_extract_frames_from_real_video(pack, tmp_path):
+    cv2 = pytest.importorskip("cv2")
+    import numpy as np
+
+    path = str(tmp_path / "clip.mp4")
+    writer = cv2.VideoWriter(path, cv2.VideoWriter_fourcc(*"mp4v"), 8, (32, 32))
+    for i in range(16):
+        frame = np.full((32, 32, 3), 255 if i == 15 else 0, dtype=np.uint8)
+        writer.write(frame)
+    writer.release()
+
+    cls = pack.NODE_CLASS_MAPPINGS["FalExtractFrames_fal"]
+    node = cls()
+    fn = getattr(node, cls.FUNCTION)
+    frames, count = fn(video=path, mode="last", n=1, max_frames=64)
+    assert count == 16
+    assert frames.shape[0] == 1
+    assert frames.mean().item() > 0.9  # last frame is white
+
+
+def test_all_util_nodes_have_tooltips(pack):
+    util_keys = [k for k, c in pack.NODE_CLASS_MAPPINGS.items() if c.CATEGORY.startswith("FAL/Utils")]
+    assert len(util_keys) == 20
+    for key in util_keys:
+        input_types = pack.NODE_CLASS_MAPPINGS[key].INPUT_TYPES()
+        for bucket in ("required", "optional"):
+            for name, spec in input_types.get(bucket, {}).items():
+                if len(spec) > 1 and isinstance(spec[1], dict):
+                    assert "tooltip" in spec[1], f"{key}.{name} missing tooltip"