fix(MediaStreamInput): ensure all frames are processed in order

- Process first frame immediately after reading for dimensions
- Use explicit index-based iteration (range(1, num_frames)) for remaining frames
- Eliminates iterator position ambiguity that could skip frame 0 or read it twice

__init__.py

@@ -70,10 +70,6 @@ from .controllers import (
     NODE_CLASS_MAPPINGS as ctrl_NODE_CLASS_MAPPINGS,
     NODE_DISPLAY_NAME_MAPPINGS as ctrl_NODE_DISPLAY_NAME_MAPPINGS,
 )
-from .rife_chunked import (
-    NODE_CLASS_MAPPINGS as rife_chunked_NODE_CLASS_MAPPINGS,
-    NODE_DISPLAY_NAME_MAPPINGS as rife_chunked_NODE_DISPLAY_NAME_MAPPINGS,
-)
 
 NODE_CLASS_MAPPINGS = dict(base_NODE_CLASS_MAPPINGS)
 NODE_DISPLAY_NAME_MAPPINGS = dict(base_NODE_DISPLAY_NAME_MAPPINGS)
@@ -87,9 +83,6 @@ NODE_DISPLAY_NAME_MAPPINGS.update(ui_NODE_DISPLAY_NAME_MAPPINGS)
 NODE_CLASS_MAPPINGS.update(ctrl_NODE_CLASS_MAPPINGS)
 NODE_DISPLAY_NAME_MAPPINGS.update(ctrl_NODE_DISPLAY_NAME_MAPPINGS)
 
-NODE_CLASS_MAPPINGS.update(rife_chunked_NODE_CLASS_MAPPINGS)
-NODE_DISPLAY_NAME_MAPPINGS.update(rife_chunked_NODE_DISPLAY_NAME_MAPPINGS)
-
 WEB_DIRECTORY = "./web"
 
 __all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS"]

media_stream.py

@@ -8,13 +8,15 @@ import imageio.v2 as imageio
 import mimetypes
 import boto3
 import json
+
+import tempfile
+import os
 from .logger import logger
 from .config.config import load_nilor_nodes_config
 
 # Load shared configuration once
 _CFG = load_nilor_nodes_config()
 
-
 # --- Node Categories ---
 category = "Nilor Nodes 👺"
 subcategories = {
@@ -54,10 +56,10 @@ class MediaStreamInput:
     CATEGORY = category + subcategories["streaming"]
 
     def download(
-        self,
-        presigned_download_url: str,
-        format: str,
-        input_name: str = "default_input",
+            self,
+            presigned_download_url: str,
+            format: str,
+            input_name: str = "default_input",
     ):
         logger.info(
             f"ℹ️\u2009 Nilor-Nodes: MediaStreamInput: Downloading from {presigned_download_url} for input '{input_name}' with format '{format}'"
@@ -94,19 +96,35 @@ class MediaStreamInput:
                 return self._process_image_batch(asset_responses)
 
             # --- Single-file download ---
-            response = requests.get(presigned_download_url, timeout=180)
-            response.raise_for_status()
-            media_bytes = response.content
-
             if format == "video":
-                return self._process_video(media_bytes)
-            elif format == "image":
-                return self._process_image(media_bytes)
+                # Stream video to temp file to avoid loading entire video into RAM
+                temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4")
+                try:
+                    logger.info(
+                        f"ℹ️\u2009 Nilor-Nodes (MediaStreamInput): Streaming video to temp file: {temp_file.name}")
+                    with requests.get(presigned_download_url, timeout=180, stream=True) as response:
+                        response.raise_for_status()
+                        for chunk in response.iter_content(chunk_size=8192):
+                            temp_file.write(chunk)
+                    temp_file.close()
+                    return self._process_video(temp_file.name)
+                finally:
+                    # Clean up temp file
+                    if os.path.exists(temp_file.name):
+                        os.unlink(temp_file.name)
             else:
-                # Should not happen if UI choices are respected
-                raise ValueError(
-                    f"🛑\u2009 Nilor-Nodes (MediaStreamInput): Unsupported format '{format}' for single media download."
-                )
+                # For images, load into memory (they're small)
+                response = requests.get(presigned_download_url, timeout=180)
+                response.raise_for_status()
+                media_bytes = response.content
+
+                if format == "image":
+                    return self._process_image(media_bytes)
+                else:
+                    # Should not happen if UI choices are respected
+                    raise ValueError(
+                        f"🛑\u2009 Nilor-Nodes (MediaStreamInput): Unsupported format '{format}' for single media download."
+                    )
 
         except requests.RequestException as e:
             logger.error(
@@ -156,33 +174,45 @@ class MediaStreamInput:
         logger.info("✅ Nilor-Nodes (MediaStreamInput): Image processing successful.")
         return (image_tensor,)
 
-    def _process_video(self, video_bytes):
-        logger.info("ℹ️\u2009 Nilor-Nodes (MediaStreamInput): Processing as video...")
-        frames = []
-        with imageio.get_reader(io.BytesIO(video_bytes), format="mp4") as reader:
-            for frame in reader:
-                # Convert frame to RGB PIL Image and then to tensor
-                pil_image = Image.fromarray(frame).convert("RGB")
-                numpy_image = np.array(pil_image).astype(np.float32) / 255.0
-                tensor_frame = torch.from_numpy(numpy_image)
-                frames.append(tensor_frame)
+    def _process_video(self, video_path):
+        logger.info(f"ℹ️\u2009 Nilor-Nodes (MediaStreamInput): Processing video from {video_path}...")
 
-        if not frames:
-            raise ValueError(
-                "🛑\u2009 Nilor-Nodes (MediaStreamInput): No frames could be read from the video."
+        # Open video to get metadata first
+        with imageio.get_reader(video_path, format="mp4") as reader:
+            # Get video metadata
+            metadata = reader.get_meta_data()
+            num_frames = reader.count_frames()
+
+            if num_frames == 0:
+                raise ValueError(
+                    "🛑\u2009 Nilor-Nodes (MediaStreamInput): No frames could be read from the video."
+                )
+
+            # Read first frame to get dimensions
+            first_frame = reader.get_data(0)
+            height, width = first_frame.shape[:2]
+
+            logger.info(
+                f"ℹ️\u2009 Nilor-Nodes (MediaStreamInput): Video has {num_frames} frames at {width}x{height}"
             )
 
-        # Pre-allocate output tensor and fill in-place, releasing each frame
-        # reference as we go. torch.stack would keep the full list alive while
-        # allocating a second equally-sized tensor, doubling peak RAM. This
-        # approach keeps peak at ~1× the final tensor size.
-        video_tensor = torch.empty(len(frames), *frames[0].shape, dtype=torch.float32)
-        for i, f in enumerate(frames):
-            video_tensor[i] = f
-            frames[i] = None  # allow GC to reclaim per-frame memory
+            # Pre-allocate tensor for all frames (N, H, W, 3)
+            video_tensor = torch.empty((num_frames, height, width, 3), dtype=torch.float32)
 
-        logging.info(
-            f"✅ Nilor-Nodes (MediaStreamInput): Video processing successful. Image Shape: {video_tensor.shape}"
+            # Process first frame (already read for dimensions)
+            pil_image = Image.fromarray(first_frame).convert("RGB")
+            numpy_image = np.array(pil_image).astype(np.float32) / 255.0
+            video_tensor[0] = torch.from_numpy(numpy_image)
+
+            # Read remaining frames by explicit index to avoid iterator position ambiguity
+            for i in range(1, num_frames):
+                frame = reader.get_data(i)
+                pil_image = Image.fromarray(frame).convert("RGB")
+                numpy_image = np.array(pil_image).astype(np.float32) / 255.0
+                video_tensor[i] = torch.from_numpy(numpy_image)
+
+        logger.info(
+            f"✅ Nilor-Nodes (MediaStreamInput): Video processing successful. Tensor shape: {video_tensor.shape}"
         )
         return (video_tensor,)
 
@@ -246,21 +276,21 @@ class MediaStreamOutput:
     CATEGORY = category + subcategories["streaming"]
 
     def upload_and_notify(
-        self,
-        images,
-        format,
-        content_id,
-        venue,
-        canvas,
-        scene,
-        presigned_upload_url,
-        job_completions_queue_url,
-        output_object_keys,
-        framerate,
-        output_name: str = "default_output",
-        prompt=None,
-        extra_pnginfo=None,
-        job_type: str | None = None,
+            self,
+            images,
+            format,
+            content_id,
+            venue,
+            canvas,
+            scene,
+            presigned_upload_url,
+            job_completions_queue_url,
+            output_object_keys,
+            framerate,
+            output_name: str = "default_output",
+            prompt=None,
+            extra_pnginfo=None,
+            job_type: str | None = None,
     ):
         if not content_id:
             raise ValueError(

rife_chunked.py

@@ -1,298 +0,0 @@
-"""RifeStreamVFI: Memory-efficient, chunked RIFE frame interpolation from a URL.
-
-Downloads a video from a presigned URL and runs RIFE N× frame interpolation in
-temporal chunks to avoid OOM on large-resolution canvases.
-
-Peak RAM per chunk (float32, chunk_size=64) vs full-video approach:
-  OLED 3840×2160  (482f):  ~18 GB/chunk   vs ~137 GB total — use 404S or Linux
-  LED_Wall 6144×1952 (482f): ~24 GB/chunk vs ~200 GB+ total — use 404S
-  Projection_Centre 5840×1072 (482f): ~14 GB/chunk vs ~108 GB total
-
-Chunking strategy:
-  - Overlap of 1 frame at chunk boundaries to ensure seamless interpolation.
-  - The first frame of each non-first chunk is the same as the last frame of
-    the previous chunk; its RIFE-duplicated counterpart in the output is
-    dropped during concatenation to avoid a double frame.
-
-Output frame count for multiplier=2, N input frames:
-  2N - 1  (identical to non-chunked RIFE VFI)
-"""
-
-from __future__ import annotations
-
-import gc
-import io
-import typing
-
-import imageio.v2 as imageio
-import numpy as np
-import requests
-import torch
-from PIL import Image
-
-from .logger import logger
-
-_DEFAULT_CHUNK_SIZE = 64
-_CATEGORY = "Nilor Nodes 👺/Streaming"
-
-
-def _get_rife_vfi_cls() -> type:
-    """Return the RIFE VFI node class from ComfyUI's registered node mappings.
-
-    Deferred import so this module can be loaded at ComfyUI startup without
-    depending on the load order of ComfyUI-Frame-Interpolation.
-
-    Returns:
-        The RIFE_VFI class.
-
-    Raises:
-        RuntimeError: If ComfyUI nodes haven't been loaded or RIFE VFI is absent.
-    """
-    try:
-        import nodes as comfy_nodes  # ComfyUI top-level nodes registry
-    except ImportError as exc:
-        raise RuntimeError(
-            "Cannot import ComfyUI 'nodes' module — is this running inside ComfyUI?"
-        ) from exc
-
-    cls = comfy_nodes.NODE_CLASS_MAPPINGS.get("RIFE VFI")
-    if cls is None:
-        raise RuntimeError(
-            "RIFE VFI node is not registered in ComfyUI NODE_CLASS_MAPPINGS. "
-            "Ensure ComfyUI-Frame-Interpolation is installed and loaded."
-        )
-    return cls
-
-
-class RifeStreamVFI:
-    """Download a video from a presigned URL and run RIFE N× interpolation in chunks.
-
-    Replaces the MediaStreamInput → RIFE VFI node pair for large canvases. Frames
-    are loaded as uint8 numpy arrays first (~3× cheaper than float32), then
-    converted to float32 one chunk at a time during inference.
-
-    Args:
-        presigned_download_url: Presigned URL to the input video.
-        input_name: Logical name used in log messages (mirrors MediaStreamInput).
-        ckpt_name: RIFE model checkpoint filename.
-        multiplier: Frame multiplier (2 = 2× frame rate, i.e. 2N-1 output frames).
-        chunk_size: Frames per chunk including 1-frame boundary overlap.
-        fast_mode: RIFE fast mode flag.
-        ensemble: RIFE ensemble flag (improves quality at minor cost).
-        scale_factor: Spatial scale for RIFE optical-flow computation.
-        dtype: Inference precision ("float32", "float16", "bfloat16").
-        batch_size: GPU batch size per RIFE forward pass.
-        clear_cache_after_n_frames: CUDA cache-clear cadence (pairs processed).
-    """
-
-    @classmethod
-    def INPUT_TYPES(cls) -> dict:
-        """Declare ComfyUI inputs for this node."""
-        return {
-            "required": {
-                "presigned_download_url": (
-                    "STRING",
-                    {"multiline": True, "default": "<auto-filled by system>"},
-                ),
-                "input_name": (
-                    "STRING",
-                    {"default": "input_video", "multiline": False},
-                ),
-                "ckpt_name": (
-                    ["rife47.pth", "rife49.pth"],
-                    {"default": "rife47.pth"},
-                ),
-                "multiplier": ("INT", {"default": 2, "min": 2, "max": 8}),
-                "chunk_size": (
-                    "INT",
-                    {
-                        "default": _DEFAULT_CHUNK_SIZE,
-                        "min": 4,
-                        "max": 512,
-                        "tooltip": (
-                            "Frames per processing chunk (1-frame overlap at boundaries). "
-                            "Lower = less peak RAM. 64 keeps 4K canvases under ~20 GB."
-                        ),
-                    },
-                ),
-                "fast_mode": ("BOOLEAN", {"default": True}),
-                "ensemble": ("BOOLEAN", {"default": True}),
-                "scale_factor": ([0.25, 0.5, 1.0, 2.0, 4.0], {"default": 1.0}),
-                "dtype": (
-                    ["float32", "float16", "bfloat16"],
-                    {"default": "float32"},
-                ),
-                "batch_size": (
-                    "INT",
-                    {
-                        "default": 1,
-                        "min": 1,
-                        "max": 64,
-                        "tooltip": "GPU batch size per RIFE forward pass.",
-                    },
-                ),
-                "clear_cache_after_n_frames": (
-                    "INT",
-                    {"default": 10, "min": 1, "max": 1000},
-                ),
-            }
-        }
-
-    RETURN_TYPES = ("IMAGE",)
-    RETURN_NAMES = ("frames",)
-    FUNCTION = "interpolate_chunked"
-    CATEGORY = _CATEGORY
-
-    def interpolate_chunked(
-        self,
-        presigned_download_url: str,
-        input_name: str = "input_video",
-        ckpt_name: str = "rife47.pth",
-        multiplier: int = 2,
-        chunk_size: int = _DEFAULT_CHUNK_SIZE,
-        fast_mode: bool = True,
-        ensemble: bool = True,
-        scale_factor: float = 1.0,
-        dtype: str = "float32",
-        batch_size: int = 1,
-        clear_cache_after_n_frames: int = 10,
-    ) -> tuple:
-        """Download a video and run chunked RIFE interpolation.
-
-        Args:
-            presigned_download_url: URL to download the source video.
-            input_name: Logical name for log messages.
-            ckpt_name: RIFE checkpoint filename.
-            multiplier: Interpolation multiplier (2 = 2× frame rate).
-            chunk_size: Frames per chunk (1-frame overlap at boundaries).
-            fast_mode: RIFE fast mode toggle.
-            ensemble: RIFE ensemble mode toggle.
-            scale_factor: Spatial scale for RIFE flow field.
-            dtype: Inference precision.
-            batch_size: GPU batch size per RIFE call.
-            clear_cache_after_n_frames: How often to clear CUDA cache.
-
-        Returns:
-            Tuple of one float32 IMAGE tensor shaped [N_out, H, W, 3].
-        """
-        logger.info(
-            "ℹ️\u2009 Nilor-Nodes (RifeStreamVFI): [%s] Starting. "
-            "chunk_size=%d multiplier=%d ckpt=%s",
-            input_name,
-            chunk_size,
-            multiplier,
-            ckpt_name,
-        )
-
-        response = requests.get(presigned_download_url, timeout=300)
-        response.raise_for_status()
-        video_bytes = response.content
-
-        # Decode all frames into uint8 numpy arrays. uint8 uses ~4× less RAM than
-        # float32 and covers the full frame count cheaply before chunk processing.
-        raw_frames: list[np.ndarray] = []
-        with imageio.get_reader(io.BytesIO(video_bytes), format="mp4") as reader:
-            for frame in reader:
-                raw_frames.append(np.asarray(Image.fromarray(frame).convert("RGB")))
-
-        del video_bytes
-
-        total_frames = len(raw_frames)
-        if total_frames < 2:
-            raise ValueError(
-                f"🛑\u2009 Nilor-Nodes (RifeStreamVFI): [{input_name}] "
-                f"Video has {total_frames} frame(s); need ≥ 2."
-            )
-
-        h, w, _ = raw_frames[0].shape
-        stride = max(1, chunk_size - 1)
-        n_chunks = max(1, (total_frames - 1 + stride - 1) // stride)
-        logger.info(
-            "ℹ️\u2009 Nilor-Nodes (RifeStreamVFI): [%s] %d frames at %dx%d — "
-            "%d chunk(s) of %d (stride %d).",
-            input_name,
-            total_frames,
-            w,
-            h,
-            n_chunks,
-            chunk_size,
-            stride,
-        )
-
-        rife_cls = _get_rife_vfi_cls()
-        rife_node = rife_cls()
-
-        output_chunks: list[torch.Tensor] = []
-        chunk_idx = 0
-        chunk_start = 0
-
-        while chunk_start < total_frames - 1:
-            chunk_end = min(chunk_start + chunk_size, total_frames)
-            chunk_raw = raw_frames[chunk_start:chunk_end]
-            n_chunk = len(chunk_raw)
-
-            logger.info(
-                "ℹ️\u2009 Nilor-Nodes (RifeStreamVFI): [%s] Chunk %d/%d — "
-                "input frames [%d..%d] (%d frames)",
-                input_name,
-                chunk_idx + 1,
-                n_chunks,
-                chunk_start,
-                chunk_end - 1,
-                n_chunk,
-            )
-
-            # Build float32 tensor for this chunk: [N, H, W, C]
-            chunk_tensor = torch.empty(n_chunk, h, w, 3, dtype=torch.float32)
-            for i, frame_np in enumerate(chunk_raw):
-                chunk_tensor[i] = torch.from_numpy(frame_np.astype(np.float32) / 255.0)
-
-            (interpolated,) = rife_node.vfi(
-                ckpt_name=ckpt_name,
-                frames=chunk_tensor,
-                multiplier=multiplier,
-                fast_mode=fast_mode,
-                ensemble=ensemble,
-                scale_factor=scale_factor,
-                dtype=dtype,
-                torch_compile=False,
-                batch_size=batch_size,
-                clear_cache_after_n_frames=clear_cache_after_n_frames,
-            )
-
-            # Skip the first output frame of non-first chunks: it duplicates the
-            # last frame of the previous chunk's output (the 1-frame overlap).
-            skip = 1 if chunk_idx > 0 else 0
-            output_chunks.append(interpolated[skip:].cpu())
-
-            del chunk_tensor, interpolated
-            gc.collect()
-
-            chunk_start += stride
-            chunk_idx += 1
-
-        total_out = sum(t.shape[0] for t in output_chunks)
-        logger.info(
-            "✅ Nilor-Nodes (RifeStreamVFI): [%s] Complete — "
-            "%d input frames → %d output frames across %d chunk(s).",
-            input_name,
-            total_frames,
-            total_out,
-            chunk_idx,
-        )
-
-        out = torch.cat(output_chunks, dim=0)
-        return (out,)
-
-
-# ---------------------------------------------------------------------------
-# ComfyUI Node Mappings
-# ---------------------------------------------------------------------------
-
-NODE_CLASS_MAPPINGS: typing.Dict[str, type] = {
-    "RifeStreamVFI": RifeStreamVFI,
-}
-
-NODE_DISPLAY_NAME_MAPPINGS: typing.Dict[str, str] = {
-    "RifeStreamVFI": "👺 RIFE Stream VFI (Chunked, from URL)",
-}

user_input.py

@@ -3,9 +3,6 @@ subcategories = {
     "io": "/IO",
 }
 
-import random
-from datetime import datetime
-
 from .controllers import CONTROLLER_HOOK
 
 
@@ -53,83 +50,6 @@ class NilorUserInput_Int:
         return (value, None)
 
 
-class NilorUserInput_Seed:
-    MAX_COMFYUI_SEED = 1125899906842624
-    SEED_RANDOM_STATE = None
-
-    @classmethod
-    def _ensure_seed_random_state(cls):
-        if cls.SEED_RANDOM_STATE is not None:
-            return
-
-        initial_random_state = random.getstate()
-        random.seed(datetime.now().timestamp())
-        cls.SEED_RANDOM_STATE = random.getstate()
-        random.setstate(initial_random_state)
-
-    @classmethod
-    def generate_random_seed(cls):
-        cls._ensure_seed_random_state()
-
-        prev_random_state = random.getstate()
-        random.setstate(cls.SEED_RANDOM_STATE)
-        seed = random.randint(0, cls.MAX_COMFYUI_SEED)
-        cls.SEED_RANDOM_STATE = random.getstate()
-        random.setstate(prev_random_state)
-        return seed
-
-    @classmethod
-    def resolve_seed(cls, value):
-        if value in (None, 0, -1):
-            return cls.generate_random_seed()
-        try:
-            return int(value) % (cls.MAX_COMFYUI_SEED + 1)
-        except (TypeError, ValueError):
-            return cls.generate_random_seed()
-
-    @classmethod
-    def INPUT_TYPES(cls):
-        return {
-            "required": {
-                "input_name": (
-                    "STRING",
-                    {"default": "my_seed_input", "multiline": False},
-                ),
-                "value": (
-                    "INT",
-                    {
-                        "default": -1,
-                        "min": -1,
-                        "max": cls.MAX_COMFYUI_SEED,
-                    },
-                ),
-            },
-            "hidden": {
-                "prompt": "PROMPT",
-                "extra_pnginfo": "EXTRA_PNGINFO",
-                "unique_id": "UNIQUE_ID",
-            },
-        }
-
-    RETURN_TYPES = ("INT", CONTROLLER_HOOK)
-    RETURN_NAMES = ("seed", "_controller_hook")
-    FUNCTION = "get_value"
-    CATEGORY = category + subcategories["io"]
-
-    @classmethod
-    def IS_CHANGED(
-        cls, input_name, value, prompt=None, extra_pnginfo=None, unique_id=None
-    ):
-        # Force node re-execution while using randomize sentinel values.
-        return cls.resolve_seed(value)
-
-    def get_value(
-        self, input_name, value, prompt=None, extra_pnginfo=None, unique_id=None
-    ):
-        value = self.resolve_seed(value)
-        return (value, None)
-
-
 class NilorUserInput_Float:
     @classmethod
     def INPUT_TYPES(cls):
@@ -177,7 +97,6 @@ class NilorUserInput_Boolean:
 NODE_CLASS_MAPPINGS = {
     "NilorUserInput_String": NilorUserInput_String,
     "NilorUserInput_Int": NilorUserInput_Int,
-    "NilorUserInput_Seed": NilorUserInput_Seed,
     "NilorUserInput_Float": NilorUserInput_Float,
     "NilorUserInput_Boolean": NilorUserInput_Boolean,
 }
@@ -185,7 +104,6 @@ NODE_CLASS_MAPPINGS = {
 NODE_DISPLAY_NAME_MAPPINGS = {
     "NilorUserInput_String": "👺 User Input (String)",
     "NilorUserInput_Int": "👺 User Input (Int)",
-    "NilorUserInput_Seed": "👺 User Input (Seed)",
     "NilorUserInput_Float": "👺 User Input (Float)",
     "NilorUserInput_Boolean": "👺 User Input (Boolean)",
 }

worker_consumer.py

@@ -29,10 +29,6 @@ from .config.config import load_nilor_nodes_config, NilorNodesConfig
 _CFG: NilorNodesConfig = load_nilor_nodes_config()
 
 
-class JobSubmissionError(Exception):
-    """Raised when a job cannot be submitted to local ComfyUI."""
-
-
 class WorkerConsumer:
     def __init__(self, cfg: NilorNodesConfig):
         self.session = get_session()
@@ -401,19 +397,7 @@ class WorkerConsumer:
                 return
 
             # Submit to ComfyUI
-            try:
-                await self._submit_job_to_comfyui(content_id, job_payload)
-            except JobSubmissionError as e:
-                logger.error(
-                    f"🛑\u2009 Nilor-Nodes (worker_consumer): Submission failed for content_id {content_id}: {e}. Message will be retried/DLQ'd."
-                )
-                await self._emit_failed_status_for_submission_error(
-                    content_id=content_id,
-                    job_payload=job_payload,
-                    error_message=str(e),
-                )
-                # Re-raise so consume_loop does not delete the message.
-                raise
+            await self._submit_job_to_comfyui(content_id, job_payload)
 
             # Cache context for subsequent status updates
             try:
@@ -434,28 +418,6 @@ class WorkerConsumer:
             # Re-raise to prevent deletion from queue if we want SQS to handle retry
             raise
 
-    async def _emit_failed_status_for_submission_error(
-        self, content_id, job_payload, error_message: str
-    ):
-        """Best-effort failed status emission for submit-time errors."""
-        policy = job_payload.get("status_policy") or {}
-        fail_status = policy.get("fail_status", "failed")
-
-        await self._send_status_update(
-            content_id,
-            fail_status,
-            job_payload.get("venue"),
-            job_payload.get("canvas"),
-            job_payload.get("scene"),
-            job_payload.get("job_type"),
-        )
-        logger.info(
-            "ℹ️\u2009 Nilor-Nodes (worker_consumer): Emitted failed status '%s' for content_id %s after submit error: %s",
-            fail_status,
-            content_id,
-            error_message,
-        )
-
     async def _submit_job_to_comfyui(self, content_id, workflow_data):
         """Submits a single job to the ComfyUI API."""
         try:
@@ -522,18 +484,15 @@ class WorkerConsumer:
             logger.error(
                 f"🛑\u2009 Nilor-Nodes (worker_consumer): Failed to submit job to ComfyUI: {e}. Message will be retried."
             )
-            raise JobSubmissionError(str(e)) from e
         except (json.JSONDecodeError, KeyError) as e:
             logger.error(
-                f"🛑\u2009 Nilor-Nodes (worker_consumer): Failed to parse ComfyUI response: {e}. Message will be retried."
+                f"🛑\u2009 Nilor-Nodes (worker_consumer): Failed to parse ComfyUI response: {e}. Discarding malformed response."
             )
-            raise JobSubmissionError(f"Malformed ComfyUI response: {e}") from e
         except Exception as e:
             logger.error(
                 f"🛑\u2009 Nilor-Nodes (worker_consumer): An unexpected error occurred while submitting job to ComfyUI: {e}",
                 exc_info=True,
             )
-            raise JobSubmissionError(str(e)) from e
 
     async def _send_status_update(
         self, content_id, status, venue=None, canvas=None, scene=None, job_type=None