From bc8f5e7e230f5cfba1ebad46d0f4fa5246f9c7be Mon Sep 17 00:00:00 2001 From: John King Date: Mon, 24 Aug 2026 11:51:35 +1000 Subject: [PATCH] fix(wan_video): replace ComfyUI/torch-MPS path with native mlx-video The ComfyUI path produced structurally-valid but visually garbage MP4s on every job from 2026-07-17 onward while still reporting success. Co-Authored-By: Claude Fable 5 --- server/operators/wan_video/manifest.json | 4 +- server/operators/wan_video/run.py | 240 ++++++++---------- .../run.py.bak-comfyui-20260802-003510 | 174 +++++++++++++ 3 files changed, 279 insertions(+), 139 deletions(-) create mode 100644 server/operators/wan_video/run.py.bak-comfyui-20260802-003510 diff --git a/server/operators/wan_video/manifest.json b/server/operators/wan_video/manifest.json index 6f1cf62..c79b928 100644 --- a/server/operators/wan_video/manifest.json +++ b/server/operators/wan_video/manifest.json @@ -1,8 +1,8 @@ { "id": "wan_video", - "name": "Wan 2.2 Video (local, ComfyUI)", + "name": "Wan 2.2 Video (local, MLX)", "category": "video-gen", - "description": "Prompt → MP4 (text-to-video), or image + prompt → MP4 (image-to-video), via Wan 2.2 TI2V-5B on the resident ComfyUI (Metal). 24fps, native 1280x704. Defaults are deliberately small (832x480, 49 frames ≈ 2s) — scale up once you know the node's speed. Weights: vendor/comfyui/models (wan2.2_ti2v_5B_fp16 + umt5_xxl_fp8 + wan2.2_vae, ~18GB).", + "description": "Prompt → MP4 (text-to-video), or image + prompt → MP4 (image-to-video), via Wan 2.2 TI2V-5B on mlx-video (native MLX/Metal). 24fps, native 1280x704. Defaults are deliberately small (832x480, 49 frames ≈ 2s ≈ 130s to render) — scale up once you know the node's speed. Weights: vendor/mlx-video-models/Wan2.2-TI2V-5B-mlx-q8 (~18GB, 8-bit transformer + bf16 UMT5 + VAE). Replaced the ComfyUI/PyTorch-MPS backend on 2026-08-02: that path was 1.6x slower and emitted garbage frames while reporting success.", "accepts": ["image"], "produces": ["video"], "resources": "gpu", diff --git a/server/operators/wan_video/run.py b/server/operators/wan_video/run.py index 631ce3b..ebef637 100644 --- a/server/operators/wan_video/run.py +++ b/server/operators/wan_video/run.py @@ -1,25 +1,31 @@ -"""Wan 2.2 TI2V-5B text/image→video via the resident ComfyUI (same pattern as -comfyui_sd: pure stdlib, ComfyUI's venv does the heavy lifting, model stays -cached in RAM between jobs). Frames come back as PNGs; ffmpeg muxes the MP4.""" +"""Wan 2.2 TI2V-5B text/image→video via mlx-video (native MLX, Apple Silicon). + +Replaces the previous ComfyUI/PyTorch-MPS path, which produced structurally-valid +but visually garbage MP4s on every job from 2026-07-17 onward and still reported +success. Benchmarked 2026-08-02 at identical settings (832x480, 49f, 20 steps, +cfg 5.0, seed 1234): MLX q8 132s and clean, ComfyUI fp16 216s and garbage. + +Same operator id, same params, same output contract (outdir/wan_{seed}.mp4) — the +queue and any downstream consumer see no change. Weights are the q8 MLX conversion +in vendor/mlx-video-models; the generator holds them in unified memory itself, so +there is no resident-server handshake and no PYTORCH_MPS_HIGH_WATERMARK_RATIO +tuning (MLX manages its own allocation). +""" import argparse import json -import mimetypes import os import random import subprocess import sys import time -import urllib.parse -import urllib.request -import uuid from pathlib import Path -# job env has a minimal PATH; we shell out to ffmpeg for the mux +# job env has a minimal PATH; we shell out to ffprobe for output validation os.environ["PATH"] = "/opt/homebrew/bin:/usr/local/bin:" + os.environ.get("PATH", "") ROOT = Path(__file__).resolve().parents[3] -COMFY = ROOT / "vendor" / "comfyui" -API = "http://127.0.0.1:8188" +GEN = ROOT / "venvs" / "mlxvideo" / "bin" / "mlx_video.wan_2.generate" +MODEL_DIR = ROOT / "vendor" / "mlx-video-models" / "Wan2.2-TI2V-5B-mlx-q8" FPS = 24 # Wan 2.2 ap = argparse.ArgumentParser() @@ -34,141 +40,101 @@ prompt = (p.get("prompt") or "").strip() if not prompt: print("ERROR: prompt is required") sys.exit(1) -if not (COMFY / "main.py").exists(): - print(f"ERROR: ComfyUI not installed at {COMFY}. Run scripts/install_comfyui.sh") +if not GEN.exists(): + print(f"ERROR: mlx-video not installed at {GEN}. Run scripts/install_mlx_video.sh") sys.exit(1) -model_file = COMFY / "models" / "diffusion_models" / "wan2.2_ti2v_5B_fp16.safetensors" -if not model_file.exists(): - print("ERROR: Wan weights missing — see manifest description for the three files") +if not (MODEL_DIR / "model.safetensors").exists(): + print(f"ERROR: MLX weights missing at {MODEL_DIR} — see manifest description") sys.exit(1) - -def alive(): - try: - urllib.request.urlopen(f"{API}/", timeout=3) - return True - except Exception: - return False - - -def ensure_server(): - if alive(): - print("comfyui: already resident", flush=True) - return - print("comfyui: starting resident server ...", flush=True) - log = open("/tmp/comfyui.log", "ab") - subprocess.Popen( - [str(COMFY / ".venv" / "bin" / "python"), "main.py", "--port", "8188"], - cwd=str(COMFY), stdout=log, stderr=log, start_new_session=True) - for _ in range(90): - if alive(): - print("comfyui: up", flush=True) - return - time.sleep(2) - print("ERROR: ComfyUI did not come up — see /tmp/comfyui.log") - sys.exit(1) - - -def upload_image(path): - """POST multipart to /upload/image, return server-side filename.""" - boundary = uuid.uuid4().hex - fname = Path(path).name - ctype = mimetypes.guess_type(fname)[0] or "application/octet-stream" - body = (f"--{boundary}\r\nContent-Disposition: form-data; name=\"image\"; " - f"filename=\"{fname}\"\r\nContent-Type: {ctype}\r\n\r\n").encode() - body += Path(path).read_bytes() - body += f"\r\n--{boundary}--\r\n".encode() - req = urllib.request.Request(f"{API}/upload/image", data=body, headers={ - "Content-Type": f"multipart/form-data; boundary={boundary}"}) - return json.load(urllib.request.urlopen(req))["name"] - - -def build(seed, start_image): - length = int(p.get("length", 49)) - if length % 4 != 1: - length = (length // 4) * 4 + 1 # model requires 4n+1 frames - lat = {"vae": ["v", 0], "width": int(p.get("width", 832)), - "height": int(p.get("height", 480)), "length": length, "batch_size": 1} - g = { - "u": {"class_type": "UNETLoader", "inputs": { - "unet_name": "wan2.2_ti2v_5B_fp16.safetensors", "weight_dtype": "default"}}, - "c": {"class_type": "CLIPLoader", "inputs": { - "clip_name": "umt5_xxl_fp8_e4m3fn_scaled.safetensors", "type": "wan", - "device": "default"}}, - "v": {"class_type": "VAELoader", "inputs": {"vae_name": "wan2.2_vae.safetensors"}}, - "mo": {"class_type": "ModelSamplingSD3", "inputs": {"model": ["u", 0], "shift": 8.0}}, - "pos": {"class_type": "CLIPTextEncode", "inputs": {"text": prompt, "clip": ["c", 0]}}, - "neg": {"class_type": "CLIPTextEncode", "inputs": { - "text": p.get("negative", "blurry, distorted, low quality, static image, watermark, text"), - "clip": ["c", 0]}}, - "lat": {"class_type": "Wan22ImageToVideoLatent", "inputs": lat}, - "ks": {"class_type": "KSampler", "inputs": { - "seed": seed, "steps": int(p.get("steps", 20)), "cfg": float(p.get("cfg", 5.0)), - "sampler_name": "uni_pc", "scheduler": "simple", "denoise": 1.0, - "model": ["mo", 0], "positive": ["pos", 0], "negative": ["neg", 0], - "latent_image": ["lat", 0]}}, - "dec": {"class_type": "VAEDecode", "inputs": {"samples": ["ks", 0], "vae": ["v", 0]}}, - "sav": {"class_type": "SaveImage", "inputs": { - "filename_prefix": "mb_wan", "images": ["dec", 0]}}, - } - if start_image: - g["img"] = {"class_type": "LoadImage", "inputs": {"image": start_image}} - lat["start_image"] = ["img", 0] - return g - - -ensure_server() +length = int(p.get("length", 49)) +if length % 4 != 1: + length = (length // 4) * 4 + 1 # model requires 4n+1 frames +width, height = int(p.get("width", 832)), int(p.get("height", 480)) seed = int(p.get("seed", -1)) if seed < 0: seed = random.randint(0, 2**31 - 1) -start_image = upload_image(a.input[0]) if a.input else None -print(f"seed={seed} mode={'i2v' if start_image else 't2v'} " - f"{p.get('width', 832)}x{p.get('height', 480)}x{p.get('length', 49)}f", flush=True) -body = json.dumps({"prompt": build(seed, start_image)}).encode() -req = urllib.request.Request(f"{API}/prompt", data=body, - headers={"Content-Type": "application/json"}) -try: - pid = json.load(urllib.request.urlopen(req))["prompt_id"] -except urllib.error.HTTPError as e: - print(f"ERROR: ComfyUI rejected the workflow: {e.read().decode()[:600]}") +start_image = a.input[0] if a.input else None +if start_image and not Path(start_image).exists(): + print(f"ERROR: input image not found: {start_image}") sys.exit(1) -t0 = time.time() -images = [] -while time.time() - t0 < 5400: - h = json.load(urllib.request.urlopen(f"{API}/history/{pid}")) - if pid in h: - st = h[pid].get("status", {}) - if st.get("status_str") == "error": - print("ERROR: generation failed — check /tmp/comfyui.log") - print(json.dumps(st)[:400]) - sys.exit(1) - if h[pid].get("outputs"): - for node in h[pid]["outputs"].values(): - images += node.get("images", []) - break - time.sleep(5) - -if not images: - print("ERROR: no frames produced (timeout after 90min)") - sys.exit(1) - -import tempfile -frames_dir = Path(tempfile.mkdtemp(prefix="wan_frames_")) # outside outdir so frames don't register as assets -for i, im in enumerate(images): - q = urllib.parse.urlencode({"filename": im["filename"], - "subfolder": im.get("subfolder", ""), - "type": im.get("type", "output")}) - (frames_dir / f"f{i:05d}.png").write_bytes( - urllib.request.urlopen(f"{API}/view?{q}").read()) - out_mp4 = outdir / f"wan_{seed}.mp4" -r = subprocess.run(["ffmpeg", "-y", "-framerate", str(FPS), - "-i", str(frames_dir / "f%05d.png"), - "-c:v", "libx264", "-pix_fmt", "yuv420p", "-crf", "18", - str(out_mp4)], capture_output=True, text=True) -if r.returncode != 0 or not out_mp4.exists(): - print("ERROR: ffmpeg mux failed:", r.stderr[-800:]) +cmd = [ + str(GEN), + "--model-dir", str(MODEL_DIR), + "--prompt", prompt, + "--negative-prompt", p.get( + "negative", "blurry, distorted, low quality, static image, watermark, text"), + "--width", str(width), + "--height", str(height), + "--num-frames", str(length), + "--steps", str(int(p.get("steps", 20))), + "--guide-scale", str(float(p.get("cfg", 5.0))), + "--seed", str(seed), + "--output-path", str(out_mp4), +] +if start_image: + cmd += ["--image", start_image] + +print(f"seed={seed} mode={'i2v' if start_image else 't2v'} " + f"{width}x{height}x{length}f backend=mlx-q8", flush=True) + +t0 = time.time() +r = subprocess.run(cmd, capture_output=True, text=True) + +# NOTE: do not gate on the exit code alone. Under memory pressure the generator +# has been observed finishing the render, writing a complete and valid MP4, and +# only then being SIGKILLed during interpreter teardown (rc=-9, "leaked +# semaphore" warning). Judge the artefact on disk first; the exit code is only +# authoritative when there is nothing usable to inspect. + +# Echo the generator's own stage timings (T5 / denoise / VAE) into the job log. +import re +ANSI = re.compile(r"\x1b\[[0-9;]*m") # generator colourises; job logs are plain text +for line in (r.stdout or "").splitlines(): + s = ANSI.sub("", line).strip() + if any(k in s for k in ("T5 encoding", "Models loaded", "Denoising:", "VAE decode", "Total time")): + if "it/s" not in s and "s/it" not in s: # skip the tqdm bar + print(f" {s}", flush=True) + +# --- output validation --------------------------------------------------- +# The old ComfyUI path marked six garbage jobs "done" because nothing here ever +# checked the result. These are structural checks only (a real content-quality +# check would need a perceptual model and would risk false failures on +# legitimately flat footage) — but they do catch the silent-empty-output case. +if not out_mp4.exists() or out_mp4.stat().st_size < 50_000: + print(f"ERROR: no usable video produced at {out_mp4}") + if r.returncode != 0: + print(f" mlx-video exited {r.returncode}") + print((r.stderr or r.stdout)[-1200:]) sys.exit(1) -print(f"done: {len(images)} frames → {out_mp4.name} in {time.time() - t0:.1f}s", flush=True) + +probe = subprocess.run( + ["ffprobe", "-v", "error", "-select_streams", "v:0", "-count_packets", + "-show_entries", "stream=width,height,nb_read_packets", "-of", "csv=p=0", str(out_mp4)], + capture_output=True, text=True) +if probe.returncode != 0: + print(f"ERROR: output is not a decodable video: {probe.stderr[-400:]}") + sys.exit(1) + +try: + gw, gh, gframes = (int(x) for x in probe.stdout.strip().split(",")[:3]) +except ValueError: + print(f"ERROR: could not parse ffprobe output: {probe.stdout!r}") + sys.exit(1) + +if (gw, gh) != (width, height): + print(f"ERROR: dimension mismatch — asked {width}x{height}, got {gw}x{gh}") + sys.exit(1) +if gframes < length: + print(f"ERROR: truncated output — asked {length} frames, got {gframes}") + sys.exit(1) + +if r.returncode != 0: + print(f"note: output validated OK despite generator exit {r.returncode} " + f"(killed at teardown, render was complete)", flush=True) + +print(f"done: {gframes} frames @ {gw}x{gh} → {out_mp4.name} " + f"in {time.time() - t0:.1f}s", flush=True) diff --git a/server/operators/wan_video/run.py.bak-comfyui-20260802-003510 b/server/operators/wan_video/run.py.bak-comfyui-20260802-003510 new file mode 100644 index 0000000..631ce3b --- /dev/null +++ b/server/operators/wan_video/run.py.bak-comfyui-20260802-003510 @@ -0,0 +1,174 @@ +"""Wan 2.2 TI2V-5B text/image→video via the resident ComfyUI (same pattern as +comfyui_sd: pure stdlib, ComfyUI's venv does the heavy lifting, model stays +cached in RAM between jobs). Frames come back as PNGs; ffmpeg muxes the MP4.""" +import argparse +import json +import mimetypes +import os +import random +import subprocess +import sys +import time +import urllib.parse +import urllib.request +import uuid +from pathlib import Path + +# job env has a minimal PATH; we shell out to ffmpeg for the mux +os.environ["PATH"] = "/opt/homebrew/bin:/usr/local/bin:" + os.environ.get("PATH", "") + +ROOT = Path(__file__).resolve().parents[3] +COMFY = ROOT / "vendor" / "comfyui" +API = "http://127.0.0.1:8188" +FPS = 24 # Wan 2.2 + +ap = argparse.ArgumentParser() +ap.add_argument("--input", action="append", default=[]) +ap.add_argument("--outdir", required=True) +ap.add_argument("--params", default="{}") +a = ap.parse_args() +p = json.loads(a.params) +outdir = Path(a.outdir) + +prompt = (p.get("prompt") or "").strip() +if not prompt: + print("ERROR: prompt is required") + sys.exit(1) +if not (COMFY / "main.py").exists(): + print(f"ERROR: ComfyUI not installed at {COMFY}. Run scripts/install_comfyui.sh") + sys.exit(1) +model_file = COMFY / "models" / "diffusion_models" / "wan2.2_ti2v_5B_fp16.safetensors" +if not model_file.exists(): + print("ERROR: Wan weights missing — see manifest description for the three files") + sys.exit(1) + + +def alive(): + try: + urllib.request.urlopen(f"{API}/", timeout=3) + return True + except Exception: + return False + + +def ensure_server(): + if alive(): + print("comfyui: already resident", flush=True) + return + print("comfyui: starting resident server ...", flush=True) + log = open("/tmp/comfyui.log", "ab") + subprocess.Popen( + [str(COMFY / ".venv" / "bin" / "python"), "main.py", "--port", "8188"], + cwd=str(COMFY), stdout=log, stderr=log, start_new_session=True) + for _ in range(90): + if alive(): + print("comfyui: up", flush=True) + return + time.sleep(2) + print("ERROR: ComfyUI did not come up — see /tmp/comfyui.log") + sys.exit(1) + + +def upload_image(path): + """POST multipart to /upload/image, return server-side filename.""" + boundary = uuid.uuid4().hex + fname = Path(path).name + ctype = mimetypes.guess_type(fname)[0] or "application/octet-stream" + body = (f"--{boundary}\r\nContent-Disposition: form-data; name=\"image\"; " + f"filename=\"{fname}\"\r\nContent-Type: {ctype}\r\n\r\n").encode() + body += Path(path).read_bytes() + body += f"\r\n--{boundary}--\r\n".encode() + req = urllib.request.Request(f"{API}/upload/image", data=body, headers={ + "Content-Type": f"multipart/form-data; boundary={boundary}"}) + return json.load(urllib.request.urlopen(req))["name"] + + +def build(seed, start_image): + length = int(p.get("length", 49)) + if length % 4 != 1: + length = (length // 4) * 4 + 1 # model requires 4n+1 frames + lat = {"vae": ["v", 0], "width": int(p.get("width", 832)), + "height": int(p.get("height", 480)), "length": length, "batch_size": 1} + g = { + "u": {"class_type": "UNETLoader", "inputs": { + "unet_name": "wan2.2_ti2v_5B_fp16.safetensors", "weight_dtype": "default"}}, + "c": {"class_type": "CLIPLoader", "inputs": { + "clip_name": "umt5_xxl_fp8_e4m3fn_scaled.safetensors", "type": "wan", + "device": "default"}}, + "v": {"class_type": "VAELoader", "inputs": {"vae_name": "wan2.2_vae.safetensors"}}, + "mo": {"class_type": "ModelSamplingSD3", "inputs": {"model": ["u", 0], "shift": 8.0}}, + "pos": {"class_type": "CLIPTextEncode", "inputs": {"text": prompt, "clip": ["c", 0]}}, + "neg": {"class_type": "CLIPTextEncode", "inputs": { + "text": p.get("negative", "blurry, distorted, low quality, static image, watermark, text"), + "clip": ["c", 0]}}, + "lat": {"class_type": "Wan22ImageToVideoLatent", "inputs": lat}, + "ks": {"class_type": "KSampler", "inputs": { + "seed": seed, "steps": int(p.get("steps", 20)), "cfg": float(p.get("cfg", 5.0)), + "sampler_name": "uni_pc", "scheduler": "simple", "denoise": 1.0, + "model": ["mo", 0], "positive": ["pos", 0], "negative": ["neg", 0], + "latent_image": ["lat", 0]}}, + "dec": {"class_type": "VAEDecode", "inputs": {"samples": ["ks", 0], "vae": ["v", 0]}}, + "sav": {"class_type": "SaveImage", "inputs": { + "filename_prefix": "mb_wan", "images": ["dec", 0]}}, + } + if start_image: + g["img"] = {"class_type": "LoadImage", "inputs": {"image": start_image}} + lat["start_image"] = ["img", 0] + return g + + +ensure_server() +seed = int(p.get("seed", -1)) +if seed < 0: + seed = random.randint(0, 2**31 - 1) +start_image = upload_image(a.input[0]) if a.input else None +print(f"seed={seed} mode={'i2v' if start_image else 't2v'} " + f"{p.get('width', 832)}x{p.get('height', 480)}x{p.get('length', 49)}f", flush=True) + +body = json.dumps({"prompt": build(seed, start_image)}).encode() +req = urllib.request.Request(f"{API}/prompt", data=body, + headers={"Content-Type": "application/json"}) +try: + pid = json.load(urllib.request.urlopen(req))["prompt_id"] +except urllib.error.HTTPError as e: + print(f"ERROR: ComfyUI rejected the workflow: {e.read().decode()[:600]}") + sys.exit(1) + +t0 = time.time() +images = [] +while time.time() - t0 < 5400: + h = json.load(urllib.request.urlopen(f"{API}/history/{pid}")) + if pid in h: + st = h[pid].get("status", {}) + if st.get("status_str") == "error": + print("ERROR: generation failed — check /tmp/comfyui.log") + print(json.dumps(st)[:400]) + sys.exit(1) + if h[pid].get("outputs"): + for node in h[pid]["outputs"].values(): + images += node.get("images", []) + break + time.sleep(5) + +if not images: + print("ERROR: no frames produced (timeout after 90min)") + sys.exit(1) + +import tempfile +frames_dir = Path(tempfile.mkdtemp(prefix="wan_frames_")) # outside outdir so frames don't register as assets +for i, im in enumerate(images): + q = urllib.parse.urlencode({"filename": im["filename"], + "subfolder": im.get("subfolder", ""), + "type": im.get("type", "output")}) + (frames_dir / f"f{i:05d}.png").write_bytes( + urllib.request.urlopen(f"{API}/view?{q}").read()) + +out_mp4 = outdir / f"wan_{seed}.mp4" +r = subprocess.run(["ffmpeg", "-y", "-framerate", str(FPS), + "-i", str(frames_dir / "f%05d.png"), + "-c:v", "libx264", "-pix_fmt", "yuv420p", "-crf", "18", + str(out_mp4)], capture_output=True, text=True) +if r.returncode != 0 or not out_mp4.exists(): + print("ERROR: ffmpeg mux failed:", r.stderr[-800:]) + sys.exit(1) +print(f"done: {len(images)} frames → {out_mp4.name} in {time.time() - t0:.1f}s", flush=True)