Upload original reference images without loading tensors

README.md

@@ -102,6 +102,12 @@ Visible video generation modes:
 
 Documented newer models such as `kling-image-o1` and `kling-video-o1` still exist in the internal capability map, but are hidden from the UI until live endpoint support is confirmed for normal accounts.
 
+## Original image uploads
+
+`Advanced Element Create` and the Kling 4.0 image/video nodes can use original JPEG/PNG files from `LoadImage`, including equal-size `Batch Images` inputs, without decoding them into tensors. Existing inputs and workflow connections stay the same. Subject images upload as files; Kling 4.0 sends the original file bytes as base64.
+
+This applies to RGB images in the current input directory. Rotated EXIF images, PNGs with EXIF, transparent or animated images, mixed-size batches, and processed images keep the existing tensor path. If another branch needs the image tensor, that branch still loads it.
+
 ## Important limitations
 
 - Native 4K video output is exposed through the video `mode` dropdown. Models that do not support `mode=4k` will be rejected by the node capability checks before submission.

py/image_files.py

@@ -0,0 +1,83 @@
+from pathlib import Path
+import re
+
+import folder_paths
+from PIL import Image
+
+
+def _plain_png(path):
+    # PNG EXIF can follow IDAT. Pillow.getexif() decodes the image to find it;
+    # skip compressed chunks instead, and leave such files on the tensor path.
+    with open(path, "rb") as source:
+        source.seek(8)
+        while header := source.read(8):
+            if len(header) != 8:
+                return False
+            length, kind = int.from_bytes(header[:4], "big"), header[4:]
+            if kind in {b"eXIf", b"acTL", b"tRNS"}:
+                return False
+            if kind == b"IEND":
+                return True
+            source.seek(length + 4, 1)
+    return False
+
+
+def _source_files(dynprompt, link, visiting):
+    if not isinstance(link, list) or len(link) != 2 or link[1] != 0:
+        return None
+    node_id = link[0]
+    if node_id in visiting or not dynprompt.has_node(node_id):
+        return None
+    node = dynprompt.get_node(node_id)
+    inputs = node["inputs"]
+    if node["class_type"] == "LoadImage" and set(inputs) == {"image"}:
+        filename = inputs["image"]
+        if not isinstance(filename, str):
+            return None
+        root = Path(folder_paths.get_input_directory()).resolve()
+        filename, annotated_root = folder_paths.annotated_filepath(filename)
+        if annotated_root is not None and Path(annotated_root).resolve() != root:
+            return None
+        path = (root / filename).resolve()
+        if Path(filename).is_absolute() or not path.is_relative_to(root):
+            raise ValueError("Reference image must be inside the current input directory.")
+        try:
+            with Image.open(path) as image:
+                if image.format not in {"JPEG", "PNG"} or image.mode != "RGB":
+                    return None
+                if image.format == "JPEG" and image.getexif().get(274, 1) != 1:
+                    return None
+                if image.format == "PNG" and not _plain_png(path):
+                    return None
+                return [(str(path), image.size)]
+        except (OSError, Image.DecompressionBombError):
+            return None
+
+    if node["class_type"] != "BatchImagesNode" or not inputs:
+        return None
+    if any(re.fullmatch(r"images\.image\d+", key) is None for key in inputs):
+        return None
+    files = []
+    for key in sorted(inputs, key=lambda key: int(key.removeprefix("images.image"))):
+        sources = _source_files(dynprompt, inputs[key], visiting | {node_id})
+        if sources is None:
+            return None
+        files.extend(sources)
+    # BatchImages resizes to its first input. Only bypass a batch that would
+    # concatenate unchanged RGB images of equal dimensions.
+    if any(size != files[0][1] for _, size in files):
+        return None
+    return files
+
+
+def original_image_files(dynprompt, unique_id, input_name):
+    if dynprompt is None or unique_id is None:
+        return None
+    link = dynprompt.get_node(unique_id)["inputs"].get(input_name)
+    sources = _source_files(dynprompt, link, set())
+    return [path for path, _ in sources] if sources is not None else None
+
+
+def required_image_inputs(dynprompt, unique_id, images):
+    return [name for name, image in images.items()
+            if image is None and original_image_files(dynprompt, unique_id, name) is None]

py/nodes.py

@@ -9,6 +9,7 @@ from .api.capabilities import CUSTOM_VOICE_LIST_TYPE, DEFAULT_IMAGE_RESOLUTIONS,
     VIDEO_GENERATION_MODES, get_image_capability, get_video_capability, validate_image_generation_request, \
     validate_video_generation_request
 from .api.exceptions import KLingAPIError
+from .image_files import original_image_files, required_image_inputs
 from .api.providers import (
     OFFICIAL_PROVIDER,
     TENCENT_VOD_PROVIDER,
@@ -3494,13 +3495,14 @@ class AdvancedCustomElementCreateNode:
                 "element_description": ("STRING", {"multiline": True, "default": ""}),
             },
             "optional": {
-                "image": ("IMAGE",),
-                "image_list": ("IMAGE",),
+                "image": ("IMAGE", {"lazy": True}),
+                "image_list": ("IMAGE", {"lazy": True}),
                 "video_url": ("STRING", {"multiline": False, "default": ""}),
                 "element_voice_id": ("STRING", {"multiline": False, "default": ""}),
                 "extra_payload_json": ("STRING", {"multiline": True, "default": ""}),
                 "reference_video_input": ("VIDEO", {"tooltip": "连接 Load Video，主体资产视频不能超过 8 秒；超长将提示更换，不自动截断。video_character 模式下优先使用此输入，否则使用 video_url。"}),
-            }
+            },
+            "hidden": {"dynprompt": "DYNPROMPT", "unique_id": "UNIQUE_ID"},
         }
 
     RETURN_TYPES = (ELEMENT_TYPE, "STRING", "STRING")
@@ -3509,6 +3511,10 @@ class AdvancedCustomElementCreateNode:
     OUTPUT_NODE = False
     CATEGORY = NODE_CATEGORY
 
+    def check_lazy_status(self, dynprompt=None, unique_id=None, **kwargs):
+        images = {name: kwargs[name] for name in ("image", "image_list") if name in kwargs}
+        return required_image_inputs(dynprompt, unique_id, images)
+
     def create(
             self,
             element_type,
@@ -3521,6 +3527,8 @@ class AdvancedCustomElementCreateNode:
             element_voice_id="",
             extra_payload_json="",
             reference_video_input=None,
+            dynprompt=None,
+            unique_id=None,
     ):
         provider = _resolve_video_provider(provider)
         generator = AdvancedCustomElements()
@@ -3541,10 +3549,15 @@ class AdvancedCustomElementCreateNode:
             generator.reference_type = "video_refer"
 
         merged_image_list = []
-        if image is not None:
-            merged_image_list.extend(_upload_image_batch_references(image))
-        if image_list is not None:
-            merged_image_list.extend(_upload_image_batch_references(image_list))
+        for name, value in (("image", image), ("image_list", image_list)):
+            if value is not None:
+                merged_image_list.extend(_upload_image_batch_references(value))
+            else:
+                files = original_image_files(dynprompt, unique_id, name)
+                if files is not None:
+                    merged_image_list.extend(_upload_file_to_temporary_media_host(path) for path in files)
+                elif dynprompt is not None and name in dynprompt.get_node(unique_id)["inputs"]:
+                    raise ValueError("Reference image changed before upload; run the workflow again.")
 
         if element_type in ("image_subject", "multi_image_subject"):
             # Kling's advanced element API requires one frontal image plus 1-3

py/nodes_v4.py

@@ -1,5 +1,6 @@
 from urllib.parse import urlsplit
 from pathlib import Path
+import base64
 import re
 import tempfile
 
@@ -12,6 +13,29 @@ from .api.video_v4 import (VIDEO_ELEMENT_TYPES, build_payload, known_duration, r
                           validate_video_metadata)
 from .nodes import (_image_batch_to_base64_list, _normalize_element_list, _normalize_voice_list,
                     _runtime_client, _upload_file_to_temporary_media_host)
+from .image_files import original_image_files, required_image_inputs
+
+
+def _image_inputs(values, lazy=False):
+    images = {name: value for name, value in values.items()
+              if name in {"first_frame", "middle_frames", "last_frame", "reference_images"}
+              or re.fullmatch(r"image_\d+", name)}
+    middle = values.get("middle_frame")
+    if lazy and isinstance(middle, tuple):
+        middle = middle[0]
+    for name, value in (middle or {}).items():
+        if lazy:
+            image, key = value
+            images[key] = image
+        else:
+            images[f"middle_frame.{name}"] = value
+    return images
+
+
+def _encode_images(images, files=None):
+    if files is not None:
+        return [base64.b64encode(Path(path).read_bytes()).decode("ascii") for path in files]
+    return _image_batch_to_base64_list(images)
 
 
 def _common_inputs():
@@ -37,24 +61,31 @@ def _frame_schema_inputs():
         io.Boolean.Input("watermark", default=False, optional=True),
         io.String.Input("callback_url", default="", optional=True),
         io.String.Input("external_task_id", default="", optional=True),
-        io.Image.Input("first_frame", optional=True, tooltip="单张首帧，所有图片按首帧→中间帧→尾帧→参考图统一编号。"),
-        io.Image.Input("middle_frames", optional=True, tooltip="旧工作流的中间帧批次，排在动态中间帧之前。"),
+        io.Image.Input("first_frame", optional=True, lazy=True, tooltip="单张首帧，所有图片按首帧→中间帧→尾帧→参考图统一编号。"),
+        io.Image.Input("middle_frames", optional=True, lazy=True, tooltip="旧工作流的中间帧批次，排在动态中间帧之前。"),
         io.Autogrow.Input("middle_frame", optional=True, template=io.Autogrow.TemplateNames(
-            io.Image.Input("frame", tooltip="中间帧，连接后自动增加接口；支持单张或批次，按接口顺序编号。"),
+            io.Image.Input("frame", lazy=True, tooltip="中间帧，连接后自动增加接口；支持单张或批次，按接口顺序编号。"),
             names=[f"middle_frame_{n}" for n in range(1, 11)], min=0)),
-        io.Image.Input("last_frame", optional=True, tooltip="单张尾帧，接在所有中间帧之后编号。"),
+        io.Image.Input("last_frame", optional=True, lazy=True, tooltip="单张尾帧，接在所有中间帧之后编号。"),
     ]
 
 
-def _frames(first_frame, middle_frames, last_frame, middle_frame=None):
+def _frames(first_frame, middle_frames, last_frame, middle_frame=None, original_files=None):
+    original_files = original_files or {}
     contents = []
-    middle = [("middle_frame", images) for name, images in sorted(
-        (middle_frame or {}).items(), key=lambda item: int(item[0].rsplit("_", 1)[1]))]
+    middle_values = dict(middle_frame or {})
+    for key in original_files:
+        if key.startswith("middle_frame."):
+            middle_values.setdefault(key.removeprefix("middle_frame."), None)
+    middle = [("middle_frame", f"middle_frame.{name}", images) for name, images in sorted(
+        middle_values.items(), key=lambda item: int(item[0].rsplit("_", 1)[1]))]
     middle_index = 0
-    for kind, images in [("first_frame", first_frame), ("middle_frame", middle_frames), *middle, ("last_frame", last_frame)]:
-        if images is None:
+    for kind, key, images in [("first_frame", "first_frame", first_frame),
+                              ("middle_frame", "middle_frames", middle_frames), *middle,
+                              ("last_frame", "last_frame", last_frame)]:
+        if images is None and key not in original_files:
             continue
-        encoded = _image_batch_to_base64_list(images)
+        encoded = _encode_images(images, original_files.get(key))
         if kind != "middle_frame" and len(encoded) != 1:
             raise ValueError(f"{kind} requires a single image; use middle_frames for a batch.")
         for value in encoded:
@@ -152,16 +183,34 @@ class Kling4Image2Video(io.ComfyNode):
             node_id="Comfyui-Kling-Wrapper Kling 4.0 Image2Video",
             category=NODE_CATEGORY + "/Kling 4.0",
             inputs=_frame_schema_inputs(),
+            hidden=[io.Hidden.dynprompt, io.Hidden.unique_id],
             outputs=[io.String.Output(display_name="url"), io.String.Output(display_name="video_id")],
         )
 
+    @classmethod
+    def check_lazy_status(cls, **kwargs):
+        hidden = cls.hidden
+        return required_image_inputs(hidden.dynprompt if hidden else None,
+                                     hidden.unique_id if hidden else None,
+                                     _image_inputs(kwargs, lazy=True))
+
     @classmethod
     def execute(cls, **kwargs):
-        return io.NodeOutput(*cls().generate(**kwargs))
+        files = {}
+        if cls.hidden is not None:
+            for name, image in _image_inputs(kwargs).items():
+                if image is None:
+                    paths = original_image_files(cls.hidden.dynprompt, cls.hidden.unique_id, name)
+                    if paths is not None:
+                        files[name] = paths
+                    elif name in cls.hidden.dynprompt.get_node(cls.hidden.unique_id)["inputs"]:
+                        raise ValueError("Reference image changed before upload; run the workflow again.")
+        return io.NodeOutput(*cls().generate(**kwargs, original_files=files))
 
     def generate(self, prompt, resolution="1080p", duration="10", first_frame=None,
-                 middle_frames=None, last_frame=None, watermark=False, callback_url="", external_task_id="", middle_frame=None):
-        prompt, contents = _unify_image_ids(prompt, _frames(first_frame, middle_frames, last_frame, middle_frame))
+                 middle_frames=None, last_frame=None, watermark=False, callback_url="", external_task_id="", middle_frame=None,
+                 original_files=None):
+        prompt, contents = _unify_image_ids(prompt, _frames(first_frame, middle_frames, last_frame, middle_frame, original_files))
         return _run("image", prompt, resolution, duration,
                     contents=contents,
                     watermark=watermark, callback_url=callback_url, external_task_id=external_task_id)
@@ -173,7 +222,7 @@ class Kling4OmniVideo(Kling4Image2Video):
         inputs = _frame_schema_inputs()
         inputs.insert(3, io.Combo.Input("aspect_ratio", options=["adaptive", "21:9", "16:9", "9:16", "1:1"], default="adaptive"))
         inputs.extend([
-            io.Image.Input("reference_images", optional=True, tooltip="旧参考图片批次，接在关键帧之后统一分配 image_N。"),
+            io.Image.Input("reference_images", optional=True, lazy=True, tooltip="旧参考图片批次，接在关键帧之后统一分配 image_N。"),
             io.Video.Input("base_video", optional=True, tooltip="待编辑视频；有此输入时为 @video_1。与 base_video_url 二选一。"),
             io.String.Input("base_video_url", optional=True, default=""),
             io.Video.Input("feature_video", optional=True, tooltip="特征参考视频；接在 base_video 后编号，无 base_video 时为 @video_1。"),
@@ -182,28 +231,32 @@ class Kling4OmniVideo(Kling4Image2Video):
             io.Custom(CUSTOM_VOICE_LIST_TYPE).Input("custom_voice_list", optional=True, tooltip="现有音色列表：@voice_1、@voice_2…"),
         ])
         for index in range(1, 11):
-            inputs.append(io.Image.Input(f"image_{index}", optional=True, tooltip="单张参考图；引用编号在所有关键帧之后，查看接口旁标签。"))
+            inputs.append(io.Image.Input(f"image_{index}", optional=True, lazy=True, tooltip="单张参考图；引用编号在所有关键帧之后，查看接口旁标签。"))
         for index in range(1, 6):
             inputs.append(io.Video.Input(f"feature_video_{index}", optional=True, tooltip="参考视频，实际 @video 编号显示在连接口旁。"))
         return io.Schema(
             node_id="Comfyui-Kling-Wrapper Kling 4.0 OmniVideo",
             category=NODE_CATEGORY + "/Kling 4.0", inputs=inputs,
+            hidden=[io.Hidden.dynprompt, io.Hidden.unique_id],
             outputs=[io.String.Output(display_name="url"), io.String.Output(display_name="video_id")],
         )
 
     def generate(self, prompt, resolution="1080p", duration="10", aspect_ratio="adaptive",
                  first_frame=None, middle_frames=None, last_frame=None, reference_images=None,
                  base_video=None, base_video_url="", feature_video=None, feature_video_urls="",
-                 element_list=None, custom_voice_list=None, watermark=False, callback_url="", external_task_id="", middle_frame=None, **dynamic_inputs):
+                 element_list=None, custom_voice_list=None, watermark=False, callback_url="", external_task_id="", middle_frame=None,
+                 original_files=None, **dynamic_inputs):
+        original_files = original_files or {}
         if base_video is not None and base_video_url.strip():
             raise ValueError("Provide base_video or base_video_url, not both.")
         urls = [_video_url(line) for line in feature_video_urls.splitlines() if line.strip()]
         base_url = _video_url(base_video_url) if base_video_url.strip() else None
-        contents = _frames(first_frame, middle_frames, last_frame, middle_frame)
+        contents = _frames(first_frame, middle_frames, last_frame, middle_frame, original_files)
         dynamic_images = []
         dynamic_videos = []
-        for name, value in dynamic_inputs.items():
-            if value is None:
+        image_files = {name: None for name in original_files if re.fullmatch(r"image_\d+", name)}
+        for name, value in (image_files | dynamic_inputs).items():
+            if value is None and name not in original_files:
                 continue
             match = re.fullmatch(r"(image|feature_video)_([1-9]\d*)", name)
             if not match:
@@ -212,15 +265,15 @@ class Kling4OmniVideo(Kling4Image2Video):
             if index > (10 if kind == "image" else 5):
                 raise ValueError(f"Kling 4.0 input index out of range: {name}")
             (dynamic_images if kind == "image" else dynamic_videos).append((index, value))
-        if dynamic_images and reference_images is not None:
+        if dynamic_images and (reference_images is not None or "reference_images" in original_files):
             raise ValueError("Use image_N inputs or legacy reference_images, not both.")
         for index, image in sorted(dynamic_images):
-            encoded = _image_batch_to_base64_list(image)
+            encoded = _encode_images(image, original_files.get(f"image_{index}"))
             if len(encoded) != 1:
                 raise ValueError(f"image_{index} requires a single image; connect each reference separately.")
             contents.append({"type": "refer_image", "url": encoded[0], "id": f"image_{index}"})
-        if reference_images is not None:
-            for index, value in enumerate(_image_batch_to_base64_list(reference_images), 1):
+        if reference_images is not None or "reference_images" in original_files:
+            for index, value in enumerate(_encode_images(reference_images, original_files.get("reference_images")), 1):
                 contents.append({"type": "refer_image", "url": value, "id": f"image_{index}"})
         videos = []
         if base_video is not None or base_url:

tests/test_original_image_files.py

@@ -0,0 +1,264 @@
+import base64
+import importlib
+import io
+import os
+from pathlib import Path
+import sys
+import tempfile
+import zlib
+from types import SimpleNamespace
+import unittest
+from unittest.mock import Mock, patch
+
+from PIL import Image
+import torch
+
+
+ROOT = Path(__file__).resolve().parents[1]
+for directory in (ROOT.parent.parent, ROOT):
+    sys.path.insert(0, str(directory))
+
+from comfy.cli_args import args
+
+args.cpu = True
+import execution
+import nodes as core_nodes
+from comfy_execution.graph import DynamicPrompt
+from comfy_extras.nodes_post_processing import BatchImagesNode
+
+kling = importlib.import_module("py.nodes")
+v4 = importlib.import_module("py.nodes_v4")
+image_files = importlib.import_module("py.image_files")
+
+
+class Collect:
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {"required": {"value": ("STRING",)}}
+
+    RETURN_TYPES = ()
+    FUNCTION = "collect"
+    OUTPUT_NODE = True
+
+    def collect(self, value):
+        return {"ui": {"value": [value]}}
+
+
+class OriginalImageTests(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(temporary.cleanup)
+        self.root = Path(temporary.name).resolve()
+        self.owner = self.root / "alice"
+        self.owner.mkdir()
+        directory = patch.object(image_files.folder_paths, "get_input_directory", return_value=str(self.owner))
+        directory.start()
+        self.addCleanup(directory.stop)
+        mapping = patch.dict(core_nodes.NODE_CLASS_MAPPINGS, {
+            "BatchImagesNode": BatchImagesNode,
+            "KlingElement": kling.AdvancedCustomElementCreateNode,
+            "KlingOmni": v4.Kling4OmniVideo,
+            "KlingImage": v4.Kling4Image2Video,
+            "Collect": Collect,
+        })
+        mapping.start()
+        self.addCleanup(mapping.stop)
+
+    def picture(self, name="photo.jpg", size=(24, 16), mode="RGB", **options):
+        path = self.owner / name
+        Image.new(mode, size, "red").save(path, **options)
+        return path
+
+    def graph(self, filename="photo.jpg"):
+        return {
+            "source": {"class_type": "LoadImage", "inputs": {"image": filename}},
+            "target": {"class_type": "KlingElement", "inputs": {"image": ["source", 0]}},
+        }
+
+    async def execute(self, prompt):
+        prompt["collect"] = {"class_type": "Collect", "inputs": {"value": ["target", 0]}}
+        server = SimpleNamespace(client_id=None, send_sync=Mock())
+        executor = execution.PromptExecutor(server, cache_type=execution.CacheType.NONE,
+                                            cache_args={"ram": 0, "ram_inactive": 0})
+        await executor.execute_async(prompt, "original-image-test", execute_outputs=["collect"])
+        self.assertTrue(executor.success, executor.status_messages)
+        return executor
+
+    def test_headers_only_and_original_bytes(self):
+        for extension in ("jpg", "png"):
+            with self.subTest(extension=extension):
+                path = self.picture(f"photo.{extension}")
+                graph = DynamicPrompt(self.graph(path.name))
+                with patch.object(Image.Image, "load", side_effect=AssertionError("must not decode")):
+                    paths = image_files.original_image_files(graph, "target", "image")
+                self.assertEqual(paths, [str(path)])
+                self.assertEqual(base64.b64decode(v4._encode_images(None, paths)[0]), path.read_bytes())
+
+    def test_rotated_transparent_and_animated_images_fall_back(self):
+        exif = Image.Exif()
+        exif[274] = 8
+        paths = [self.picture("rotated.jpg", exif=exif),
+                 self.picture("alpha.png", mode="RGBA"),
+                 self.picture("key.png", transparency=(255, 0, 0))]
+        animated = self.owner / "animated.png"
+        Image.new("RGB", (24, 16), "red").save(
+            animated, save_all=True, append_images=[Image.new("RGB", (24, 16), "blue")])
+        paths.append(animated)
+        for path in paths:
+            with self.subTest(path=path.name):
+                self.assertIsNone(image_files.original_image_files(DynamicPrompt(self.graph(path.name)), "target", "image"))
+
+    def test_png_exif_after_pixel_data_is_detected_without_decoding(self):
+        path = self.picture("late-exif.png")
+        exif = Image.Exif()
+        exif[274] = 8
+        data = exif.tobytes()[6:]
+        chunk = b"eXIf" + data
+        chunk = len(data).to_bytes(4, "big") + chunk + zlib.crc32(chunk).to_bytes(4, "big")
+        original = path.read_bytes()
+        path.write_bytes(original[:-12] + chunk + original[-12:])
+        with patch.object(Image.Image, "load", side_effect=AssertionError("must not decode")):
+            self.assertIsNone(image_files.original_image_files(DynamicPrompt(self.graph(path.name)), "target", "image"))
+
+    def test_batch_order_duplicates_and_mixed_sizes(self):
+        first = self.picture("a.png")
+        second = self.picture("b.jpg")
+        graph = self.graph("a.png")
+        graph["second"] = {"class_type": "LoadImage", "inputs": {"image": "b.jpg"}}
+        graph["batch"] = {"class_type": "BatchImagesNode", "inputs": {
+            "images.image2": ["second", 0], "images.image0": ["source", 0], "images.image1": ["second", 0]}}
+        graph["target"]["inputs"]["image"] = ["batch", 0]
+        self.assertEqual(image_files.original_image_files(DynamicPrompt(graph), "target", "image"),
+                         [str(first), str(second), str(second)])
+        self.picture("b.jpg", size=(12, 8))
+        self.assertIsNone(image_files.original_image_files(DynamicPrompt(graph), "target", "image"))
+
+    def test_processed_missing_and_mask_inputs_fall_back(self):
+        self.picture()
+        for source in (
+            {"class_type": "ImageScale", "inputs": {"image": ["source", 0]}},
+            {"class_type": "LoadImage", "inputs": {"image": "missing.jpg"}},
+        ):
+            graph = self.graph()
+            graph["processed"] = source
+            graph["target"]["inputs"]["image"] = ["processed", 0]
+            self.assertIsNone(image_files.original_image_files(DynamicPrompt(graph), "target", "image"))
+        graph = self.graph()
+        graph["target"]["inputs"]["image"] = ["source", 1]
+        self.assertIsNone(image_files.original_image_files(DynamicPrompt(graph), "target", "image"))
+
+    def test_current_account_directory_is_enforced(self):
+        other = self.root / "bob"
+        other.mkdir()
+        path = other / "private.jpg"
+        Image.new("RGB", (24, 16)).save(path)
+        for filename in ("../bob/private.jpg", str(path)):
+            with self.subTest(filename=filename), self.assertRaisesRegex(ValueError, "current input directory"):
+                image_files.original_image_files(DynamicPrompt(self.graph(filename)), "target", "image")
+
+    def test_symlink_cannot_escape_current_account(self):
+        outside = self.root / "outside.jpg"
+        Image.new("RGB", (24, 16)).save(outside)
+        try:
+            os.symlink(outside, self.owner / "link.jpg")
+        except OSError as error:
+            self.skipTest(str(error))
+        with self.assertRaisesRegex(ValueError, "current input directory"):
+            image_files.original_image_files(DynamicPrompt(self.graph("link.jpg")), "target", "image")
+
+    async def test_subject_creation_skips_load_and_batch_and_uploads_original_files(self):
+        path = self.picture()
+        graph = self.graph()
+        graph["batch"] = {"class_type": "BatchImagesNode", "inputs": {
+            "images.image0": ["source", 0], "images.image1": ["source", 0]}}
+        graph["target"]["inputs"].update(element_type="multi_image_subject", element_name="subject",
+            element_description="subject description", image_list=["batch", 0])
+        uploaded = []
+
+        def upload(filename):
+            uploaded.append(Path(filename).read_bytes())
+            return "https://example.com/original.jpg"
+
+        provider = Mock()
+        provider.create_element.return_value = SimpleNamespace(
+            element={"element_id": "element", "reference_type": "image_refer"}, element_id="element")
+        with patch.object(core_nodes.LoadImage, "load_image", side_effect=AssertionError("LoadImage executed")), \
+             patch.object(BatchImagesNode, "execute", side_effect=AssertionError("Batch executed")), \
+             patch.object(kling, "_resolve_video_provider", return_value=kling.TENCENT_VOD_PROVIDER), \
+             patch.object(kling, "_upload_file_to_temporary_media_host", side_effect=upload), \
+             patch.object(kling, "_video_provider") as backend:
+            backend.return_value.__enter__.return_value = provider
+            await self.execute(graph)
+        self.assertEqual(uploaded, [path.read_bytes()] * 3)
+        self.assertTrue(path.exists())
+
+    async def test_omni_skips_sources_and_keeps_dynamic_frame_order(self):
+        path = self.picture()
+        graph = self.graph()
+        graph["target"] = {"class_type": "KlingOmni", "inputs": {
+            "prompt": "@image_1", "resolution": "720p", "duration": "10", "aspect_ratio": "adaptive",
+            "middle_frame.middle_frame_2": ["source", 0], "middle_frame.middle_frame_1": ["source", 0],
+            "image_1": ["source", 0]}}
+        with patch.object(core_nodes.LoadImage, "load_image", side_effect=AssertionError("LoadImage executed")), \
+             patch.object(v4, "_run", return_value=("url", "id")) as run:
+            await self.execute(graph)
+        contents = run.call_args.args[5]
+        self.assertEqual([item["type"] for item in contents], ["middle_frame", "middle_frame", "refer_image"])
+        self.assertEqual([item["id"] for item in contents], ["image_1", "image_2", "image_3"])
+        self.assertTrue(all(base64.b64decode(item["url"]) == path.read_bytes() for item in contents))
+
+    async def test_rotated_photo_executes_existing_loader(self):
+        exif = Image.Exif()
+        exif[274] = 8
+        self.picture(exif=exif)
+        graph = self.graph()
+        graph["target"] = {"class_type": "KlingImage", "inputs": {
+            "prompt": "photo", "resolution": "720p", "duration": "10", "first_frame": ["source", 0]}}
+        with patch.object(core_nodes.LoadImage, "load_image", autospec=True, return_value=(torch.ones(1, 24, 16, 3), None)) as load, \
+             patch.object(v4, "_run", return_value=("url", "id")) as run:
+            await self.execute(graph)
+        load.assert_called_once()
+        with Image.open(io.BytesIO(base64.b64decode(run.call_args.kwargs["contents"][0]["url"]))) as image:
+            self.assertEqual(image.size, (16, 24))
+
+    async def test_resize_in_dynamic_middle_frame_uses_processed_pixels(self):
+        self.picture()
+        graph = self.graph()
+        graph["resize"] = {"class_type": "ImageScale", "inputs": {
+            "image": ["source", 0], "upscale_method": "nearest-exact", "width": 8, "height": 8, "crop": "disabled"}}
+        graph["target"] = {"class_type": "KlingImage", "inputs": {
+            "prompt": "photo", "resolution": "720p", "duration": "10", "middle_frame.middle_frame_1": ["resize", 0]}}
+        with patch.object(core_nodes.LoadImage, "load_image", autospec=True, return_value=(torch.ones(1, 16, 24, 3), None)) as load, \
+             patch.object(v4, "_run", return_value=("url", "id")) as run:
+            await self.execute(graph)
+        load.assert_called_once()
+        with Image.open(io.BytesIO(base64.b64decode(run.call_args.kwargs["contents"][0]["url"]))) as image:
+            self.assertEqual(image.size, (8, 8))
+
+    async def test_mixed_size_batch_keeps_existing_resize_behavior(self):
+        self.picture("a.jpg")
+        self.picture("b.jpg", size=(12, 8))
+        graph = self.graph("a.jpg")
+        graph["second"] = {"class_type": "LoadImage", "inputs": {"image": "b.jpg"}}
+        graph["batch"] = {"class_type": "BatchImagesNode", "inputs": {
+            "images.image0": ["source", 0], "images.image1": ["second", 0]}}
+        graph["target"] = {"class_type": "KlingOmni", "inputs": {
+            "prompt": "photo", "resolution": "720p", "duration": "10", "aspect_ratio": "adaptive",
+            "reference_images": ["batch", 0]}}
+
+        def load_image(instance, image):
+            return torch.ones(1, 16, 24, 3) if image == "a.jpg" else torch.ones(1, 8, 12, 3), None
+
+        with patch.object(core_nodes.LoadImage, "load_image", autospec=True, side_effect=load_image) as load, \
+             patch.object(v4, "_run", return_value=("url", "id")) as run:
+            await self.execute(graph)
+        self.assertEqual(load.call_count, 2)
+        contents = run.call_args.args[5]
+        self.assertEqual(len(contents), 2)
+        for item in contents:
+            with Image.open(io.BytesIO(base64.b64decode(item["url"]))) as image:
+                self.assertEqual(image.size, (24, 16))
+
+
+if __name__ == "__main__":
+    unittest.main()