audio/video/text operator suite: stt_local, lipsync_local, voice_clone_local, wan_video, llm_local

- stt_local: Whisper large-v3-turbo MLX, transcript + timestamped json
- lipsync_local: Rhubarb visemes (CPU lane), dialog-guided
- voice_clone_local: Chatterbox on MPS (needs setuptools<81 for perth/pkg_resources)
- wan_video: Wan 2.2 TI2V-5B t2v/i2v via resident ComfyUI, frames->mp4
- llm_local: Qwen3-30B-A3B-4bit dialogue writer (primary-only)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

scripts/install_rhubarb.sh

@@ -0,0 +1,16 @@
+#!/usr/bin/env bash
+# Install Rhubarb Lip Sync (audio → viseme timings) into vendor/rhubarb.
+# Official binary is x86_64 — runs via Rosetta on Apple Silicon Macs.
+set -euo pipefail
+cd "$(dirname "$0")/.."
+V=1.14.0
+[ -x vendor/rhubarb/rhubarb ] && { echo "already installed: $(vendor/rhubarb/rhubarb --version)"; exit 0; }
+mkdir -p vendor
+curl -fsSL -o /tmp/rhubarb.zip \
+  "https://github.com/DanielSWolf/rhubarb-lip-sync/releases/download/v${V}/Rhubarb-Lip-Sync-${V}-macOS.zip"
+unzip -q -o /tmp/rhubarb.zip -d vendor/
+rm -rf vendor/rhubarb
+mv "vendor/Rhubarb-Lip-Sync-${V}-macOS" vendor/rhubarb
+xattr -dr com.apple.quarantine vendor/rhubarb 2>/dev/null || true
+rm -f /tmp/rhubarb.zip
+vendor/rhubarb/rhubarb --version

server/operators/lipsync_local/manifest.json

@@ -0,0 +1,17 @@
+{
+  "id": "lipsync_local",
+  "name": "Lip Sync (local, Rhubarb)",
+  "category": "audio",
+  "description": "Speech WAV → mouth-shape timing JSON via Rhubarb Lip Sync (fully offline, CPU). Output is timed Preston Blair visemes (A-F + optional G/H/X) — map each shape once to a blendshape/pose on a rigged head and every voice line animates itself. Pass the known transcript as 'dialog' for noticeably better accuracy (we always have it for tts_local lines). Input must be WAV/OGG. Install: scripts/install_rhubarb.sh",
+  "accepts": ["audio"],
+  "produces": ["data"],
+  "resources": "cpu",
+  "entry": "run.py",
+  "params_schema": {
+    "type": "object",
+    "properties": {
+      "dialog": {"type": "string", "default": "", "description": "The spoken text, if known — improves phoneme alignment"},
+      "shapes": {"type": "string", "enum": ["extended", "basic"], "default": "extended", "description": "extended = A-F+GHX (8 shapes), basic = A-F (6)"}
+    }
+  }
+}

server/operators/lipsync_local/run.py

@@ -0,0 +1,49 @@
+import argparse
+import json
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+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)
+
+if not a.input:
+    print("ERROR: needs one WAV/OGG input")
+    sys.exit(1)
+
+root = Path(__file__).resolve().parents[3]
+rhubarb = root / "vendor" / "rhubarb" / "rhubarb"
+if not rhubarb.exists():
+    print("ERROR: rhubarb not installed — run scripts/install_rhubarb.sh")
+    sys.exit(1)
+
+outdir = Path(a.outdir)
+out_json = outdir / (Path(a.input[0]).stem + ".visemes.json")
+cmd = [str(rhubarb), "-f", "json", "-o", str(out_json), a.input[0]]
+if p.get("shapes", "extended") == "basic":
+    cmd += ["--extendedShapes", ""]
+
+dialog = (p.get("dialog") or "").strip()
+if dialog:
+    tmp = outdir / "work"  # subdir: only top-level outdir files register as assets
+    tmp.mkdir(exist_ok=True)
+    dialog_file = tmp / "dialog.txt"
+    dialog_file.write_text(dialog + "\n")
+    cmd += ["--dialogFile", str(dialog_file)]
+
+t0 = time.time()
+r = subprocess.run(cmd, capture_output=True, text=True)
+if r.stderr:
+    print(r.stderr[-1500:], file=sys.stderr)
+if r.returncode != 0 or not out_json.exists():
+    print("ERROR: rhubarb failed")
+    sys.exit(1)
+
+cues = json.loads(out_json.read_text())["mouthCues"]
+print(f"done in {time.time() - t0:.1f}s: {len(cues)} mouth cues, "
+      f"{cues[-1]['end'] if cues else 0:.2f}s")

server/operators/llm_local/manifest.json

@@ -0,0 +1,20 @@
+{
+  "id": "llm_local",
+  "name": "LLM (local, Qwen3-30B)",
+  "category": "text",
+  "description": "Prompt → text via Qwen3-30B-A3B-4bit on MLX (~17GB, MoE with 3B active params = fast on big unified memory). Batch NPC dialogue, item descriptions, lore — pipe straight into tts_local. For quick/cheap calls use the m4pro Ollama endpoint instead; this is the heavyweight. Primary-node only.",
+  "accepts": [],
+  "produces": ["text"],
+  "resources": "gpu",
+  "entry": "run.py",
+  "python": "venvs/llm/bin/python",
+  "params_schema": {
+    "type": "object",
+    "properties": {
+      "prompt": {"type": "string", "default": "", "description": "The user prompt"},
+      "system": {"type": "string", "default": "", "description": "Optional system prompt (character sheet, style rules)"},
+      "max_tokens": {"type": "integer", "default": 1024, "minimum": 16, "maximum": 8192},
+      "temperature": {"type": "number", "default": 0.7, "minimum": 0, "maximum": 2}
+    }
+  }
+}

server/operators/llm_local/run.py

@@ -0,0 +1,43 @@
+import argparse
+import json
+import re
+import sys
+import time
+from pathlib import Path
+
+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)
+
+prompt = (p.get("prompt") or "").strip()
+if not prompt:
+    print("ERROR: prompt is required")
+    sys.exit(1)
+
+from mlx_lm import load, generate  # noqa: E402
+from mlx_lm.sample_utils import make_sampler  # noqa: E402
+
+MODEL = "mlx-community/Qwen3-30B-A3B-4bit"
+t0 = time.time()
+model, tokenizer = load(MODEL)
+print(f"model loaded in {time.time() - t0:.1f}s", flush=True)
+
+messages = []
+if (p.get("system") or "").strip():
+    messages.append({"role": "system", "content": p["system"].strip()})
+messages.append({"role": "user", "content": prompt})
+chat = tokenizer.apply_chat_template(
+    messages, add_generation_prompt=True, tokenize=False, enable_thinking=False)
+
+t1 = time.time()
+text = generate(model, tokenizer, prompt=chat,
+                max_tokens=int(p.get("max_tokens", 1024)),
+                sampler=make_sampler(temp=float(p.get("temperature", 0.7))))
+text = re.sub(r"<think>.*?</think>", "", text, flags=re.S).strip()
+
+out = Path(a.outdir) / "llm_output.txt"
+out.write_text(text + "\n")
+print(f"done in {time.time() - t1:.1f}s, {len(text)} chars:\n{text[:400]}", flush=True)

server/operators/stt_local/manifest.json

@@ -0,0 +1,17 @@
+{
+  "id": "stt_local",
+  "name": "Transcribe (local, Whisper)",
+  "category": "audio",
+  "description": "Audio → transcript (.txt + timestamped .json) entirely on this Mac via Whisper large-v3-turbo on MLX. Subtitles, VO QA round-trips (does the generated line transcribe back correctly?), dictation. Needs ffmpeg on PATH for non-wav inputs.",
+  "accepts": ["audio"],
+  "produces": ["text"],
+  "resources": "gpu",
+  "entry": "run.py",
+  "python": "venvs/whisper/bin/python",
+  "params_schema": {
+    "type": "object",
+    "properties": {
+      "language": {"type": "string", "default": "en", "description": "ISO language code; empty = autodetect"}
+    }
+  }
+}

server/operators/stt_local/run.py

@@ -0,0 +1,36 @@
+import argparse
+import json
+import sys
+import time
+from pathlib import Path
+
+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)
+
+if not a.input:
+    print("ERROR: needs one audio input")
+    sys.exit(1)
+
+import mlx_whisper  # noqa: E402 — import after arg errors so bad calls fail fast
+
+t0 = time.time()
+result = mlx_whisper.transcribe(
+    a.input[0],
+    path_or_hf_repo="mlx-community/whisper-large-v3-turbo",
+    language=p.get("language") or None,
+)
+text = result["text"].strip()
+outdir = Path(a.outdir)
+stem = Path(a.input[0]).stem
+(outdir / f"{stem}.txt").write_text(text + "\n")
+(outdir / f"{stem}.whisper.json").write_text(json.dumps({
+    "text": text,
+    "language": result.get("language"),
+    "segments": [{"start": s["start"], "end": s["end"], "text": s["text"].strip()}
+                 for s in result["segments"]],
+}, indent=2))
+print(f"done in {time.time() - t0:.1f}s: {text[:150]}")

server/operators/voice_clone_local/manifest.json

@@ -0,0 +1,19 @@
+{
+  "id": "voice_clone_local",
+  "name": "Voice Clone (local, Chatterbox)",
+  "category": "audio",
+  "description": "Reference voice WAV (~10s of clean speech) + text → that voice speaking the line. Chatterbox (Resemble AI, MIT) on MPS, with an emotion 'exaggeration' knob (0.3 flat … 0.7 dramatic). Record one growl per monster, save it in character_kit/voices/, and every line that character ever speaks comes out in-voice. Output is watermarked as AI audio (Perth, inaudible) by the model vendor. Slower than tts_local — use Kokoro for stock voices.",
+  "accepts": ["audio"],
+  "produces": ["audio"],
+  "resources": "gpu",
+  "entry": "run.py",
+  "python": "venvs/chatterbox/bin/python",
+  "params_schema": {
+    "type": "object",
+    "properties": {
+      "text": {"type": "string", "default": "", "description": "The line to speak"},
+      "exaggeration": {"type": "number", "default": 0.5, "minimum": 0.25, "maximum": 1.0, "description": "Emotion intensity (0.5 = neutral-ish)"},
+      "cfg_weight": {"type": "number", "default": 0.5, "minimum": 0.2, "maximum": 1.0, "description": "Lower = slower, more deliberate delivery"}
+    }
+  }
+}

server/operators/voice_clone_local/run.py

@@ -0,0 +1,44 @@
+import argparse
+import json
+import os
+import sys
+import time
+from pathlib import Path
+
+os.environ.setdefault("TQDM_DISABLE", "1")  # keep sampling bars out of job logs
+
+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)
+
+text = (p.get("text") or "").strip()
+if not text:
+    print("ERROR: text is required")
+    sys.exit(1)
+if not a.input:
+    print("ERROR: needs a reference voice WAV as input")
+    sys.exit(1)
+
+import torch  # noqa: E402
+import torchaudio  # noqa: E402
+from chatterbox.tts import ChatterboxTTS  # noqa: E402
+
+device = "mps" if torch.backends.mps.is_available() else "cpu"
+t0 = time.time()
+model = ChatterboxTTS.from_pretrained(device=device)
+print(f"model loaded on {device} in {time.time() - t0:.1f}s", flush=True)
+
+t1 = time.time()
+wav = model.generate(
+    text,
+    audio_prompt_path=a.input[0],
+    exaggeration=float(p.get("exaggeration", 0.5)),
+    cfg_weight=float(p.get("cfg_weight", 0.5)),
+)
+out = Path(a.outdir) / f"clone_{Path(a.input[0]).stem}.wav"
+torchaudio.save(str(out), wav, model.sr)
+dur = wav.shape[-1] / model.sr
+print(f"done in {time.time() - t1:.1f}s: {out.name} ({dur:.1f}s)", flush=True)

server/operators/wan_video/manifest.json

@@ -0,0 +1,23 @@
+{
+  "id": "wan_video",
+  "name": "Wan 2.2 Video (local, ComfyUI)",
+  "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).",
+  "accepts": ["image"],
+  "produces": ["video"],
+  "resources": "gpu",
+  "entry": "run.py",
+  "params_schema": {
+    "type": "object",
+    "properties": {
+      "prompt": {"type": "string", "default": "", "description": "What happens in the shot"},
+      "negative": {"type": "string", "default": "blurry, distorted, low quality, static image, watermark, text", "description": "What to avoid"},
+      "width": {"type": "integer", "default": 832, "minimum": 256, "maximum": 1280},
+      "height": {"type": "integer", "default": 480, "minimum": 256, "maximum": 1280},
+      "length": {"type": "integer", "default": 49, "minimum": 5, "maximum": 121, "description": "Frames, must be 4n+1 (49≈2s, 121≈5s at 24fps)"},
+      "steps": {"type": "integer", "default": 20, "minimum": 4, "maximum": 50},
+      "cfg": {"type": "number", "default": 5.0, "minimum": 1, "maximum": 15},
+      "seed": {"type": "integer", "default": -1, "description": "-1 = random"}
+    }
+  }
+}

server/operators/wan_video/run.py

@@ -0,0 +1,170 @@
+"""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 random
+import subprocess
+import sys
+import time
+import urllib.parse
+import urllib.request
+import uuid
+from pathlib import 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)
+
+frames_dir = outdir / "work"  # subdir: keep raw frames out of asset registration
+frames_dir.mkdir(exist_ok=True)
+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)