video: run MiniMax-H3 in the engine like any other model, worker as fallback

H3 doesn't need a permanent sidecar. Model work already happens in a supervised,
per-GPU engine subprocess with its own CUDA context, VRAM manager and swap gate —
so H3 belongs there, registered with the model manager like a Wan or LTX
pipeline, and pinned to a CUDA engine.

What actually forced the worker was the version floor, not the architecture:
diffusers ships H3 as Modular blocks only (0.40+), and this venv is on 0.38.
So the route is now chosen by capability, not hardcoded — engine_supports_h3()
probes for MiniMaxH3ModularPipeline, and `in_process` in the model config forces
either side. The same build works before and after a diffusers upgrade, and the
worker stays as the fallback while H3 is unproven (an in-engine OOM or device
assert poisons the whole engine's context; a subprocess crash doesn't).

_load_h3_pipeline() builds the modular pipeline with ONE workflow (loading with
none pulls both 61.7GB partitions), group-offloads the big modules per component
since a ModularPipeline has no pipeline-level offload helper, and registers it
with record_vram_delta so eviction sees its real footprint. A workflow change
unloads first — the two partitions can't be co-resident.

The checkpoint contracts now live in tools/h3_common.py, loaded by path from the
engine and imported as a sibling module inside the isolated venv, so both routes
snap frames, round the canvas, pick the workflow and mux the audio identically.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0182DhR1eFNPQyGmDrpHbedY

codai/api/h3_worker.py

@@ -62,6 +62,58 @@ _services: dict = {}          # model -> {"proc","port","url"}
 _bootstrapped = False
 
 
+# ── shared rules + in-engine capability ──────────────────────────────────────
+
+_rules = None
+
+
+def rules():
+    """The H3 checkpoint contracts, shared with the isolated worker.
+
+    Loaded by path from tools/h3_common.py: the isolated venv imports it as a
+    sibling module, and the engine can't reach it through the `codai.api` package
+    (importing that pulls the whole FastAPI app into the worker's venv).
+    """
+    global _rules
+    if _rules is None:
+        import importlib.util
+        path = _REPO_ROOT / "tools" / "h3_common.py"
+        spec = importlib.util.spec_from_file_location("h3_common", path)
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        _rules = module
+    return _rules
+
+
+def engine_supports_h3() -> bool:
+    """True when THIS process's diffusers carries the H3 modular pipeline.
+
+    H3 needs diffusers >= 0.40. When the engine has it, H3 loads in-engine like
+    every other video model (normal eviction, offload and VRAM accounting); when
+    it doesn't, the isolated worker is used instead. Auto-detected so the same
+    build works either side of a diffusers upgrade.
+    """
+    try:
+        from diffusers import MiniMaxH3ModularPipeline  # noqa: F401
+        return True
+    except Exception:
+        return False
+
+
+def use_worker(config: dict = None) -> bool:
+    """Whether this model should go through the isolated worker.
+
+    ``in_process`` in the model config forces the choice either way; otherwise it
+    follows what the engine's diffusers can actually do.
+    """
+    cfg = config or {}
+    if "in_process" in cfg:
+        return not bool(cfg.get("in_process"))
+    if str(cfg.get("backend") or "").lower() in ("h3-worker", "h3_worker"):
+        return True
+    return not engine_supports_h3()
+
+
 # ── detection ─────────────────────────────────────────────────────────────────
 
 def is_h3_model(model_name: str = "", config: dict = None) -> bool:

docs/minimax-h3.md

@@ -2,19 +2,35 @@
 
 H3 is a 33B flow-matching DiT that denoises **video and audio in one packed
 sequence** — every generation comes back as an mp4 that already has a synchronised
-soundtrack, not a dub. It is served through an isolated worker rather than the
-normal diffusers loader, for one blunt reason: diffusers ships H3 as Modular
-Diffusers blocks only (`MiniMaxH3Blocks` / `MiniMaxH3ModularPipeline`, 0.40+),
-there is no `DiffusionPipeline` half, and the main venv is on diffusers 0.38.
+soundtrack, not a dub.
+
+It runs **in the engine, like every other video model**, whenever that engine's
+diffusers can do H3 (>= 0.40): the pipeline is registered with the model manager,
+so eviction, VRAM accounting, the RAM cap and the GPU swap gate all treat it
+normally. Pin it to a CUDA engine (`"engine": "nvidia"` in models.json) — a 33B
+bf16 DiT is not going to run on the Vulkan engine.
+
+When the engine's diffusers is older, the same request is served by an isolated
+venv worker instead, so the feature works either side of a diffusers upgrade:
 
 ```
-/v1/video/generations ──► codai/api/video.py::_generate_h3
-                            └─► codai/api/h3_worker.py   (venv + process + eviction)
-                                  └─► tools/h3_service.py  (isolated venv, HTTP)
-                                        └─► ModularPipeline.from_pretrained(..., workflow=…)
+/v1/video/generations ──► video.py::_generate_h3
+                            ├─ in-engine  (diffusers >= 0.40)
+                            │    _load_h3_pipeline → ModularPipeline(workflow=…)
+                            └─ worker     (older diffusers)
+                                 h3_worker → tools/h3_service.py (isolated venv)
 ```
 
-Same shape as the pyannote, NeMo-Canary and vLLM workers.
+The choice is automatic (`h3_worker.engine_supports_h3()`), and `"in_process":
+true|false` in the model config forces either side. Both paths share the
+checkpoint contracts in `tools/h3_common.py` — the frame/canvas arithmetic, the
+workflow choice and the audio mux are identical, so the two routes produce the
+same thing.
+
+Why H3 needs its own branch at all, in either mode: diffusers ships it as Modular
+Diffusers blocks only (`MiniMaxH3Blocks` / `MiniMaxH3ModularPipeline`), with no
+`DiffusionPipeline` half — so `_detect_pipeline_class()` /
+`PClass.from_pretrained` can't express it.
 
 ## Configure a model
 
@@ -39,11 +55,15 @@ Add an entry to `models.json`. Detection is by name (anything containing
 | `offload_strategy` | `group` (default), `leaf` (smaller groups, slower), or `""` for straight-to-GPU. The diffusers CPU-offload strategies have no modular equivalent and are normalised to `group`. |
 | `device_map` | per-component device map (`auto`, `balanced`) instead of group offload |
 | `used_vram_gb` | reserved before start, and what eviction frees. Default 22. |
-| `h3_venv` | override the isolated venv location |
+| `in_process` | force in-engine (`true`) or the isolated worker (`false`); default: whatever the engine's diffusers supports |
+| `h3_venv` | override the isolated venv location (worker mode only) |
+| `engine` | pin the owning engine — use your CUDA one |
 | `gpu_device` | pins `CUDA_VISIBLE_DEVICES` for the worker |
 
-The venv builds itself on first use from `requirements-h3.txt` (several GB) under
-`/opt/coderai/h3_venv`, the `/cache` mount, or `~/.coderai/h3_venv`.
+In worker mode the venv builds itself on first use from `requirements-h3.txt`
+(several GB) under `/opt/coderai/h3_venv`, the `/cache` mount, or
+`~/.coderai/h3_venv`. In-engine mode needs nothing extra — just diffusers >= 0.40
+(which also wants `huggingface_hub >= 1.23`) in the engine's own venv.
 
 ## What the request translation does
 
@@ -80,8 +100,12 @@ the audio loss zeroed (stills carry no audio); or 50–200 clips of 3–15 s at
 ## Operational notes
 
 - **Weights are ~62 GB per partition.** Budget disk and first-run download time.
-- The worker is registered as an evictable, VRAM-tracked model; eviction stops the
-  whole subprocess, which is the only way the transformer's pages actually go back.
+- Both modes register as evictable, VRAM-tracked models. In-engine, eviction drops
+  the pipeline like any other; in worker mode it stops the whole subprocess, which
+  is the only way that process's pages actually go back.
+- In-engine, an H3 OOM or device-side assert poisons the engine's CUDA context and
+  takes down every model on that engine — the reason the worker mode exists as a
+  fallback while H3 is unproven.
 - Switching workflow within one model reloads the other partition — batch requests
   by workflow if you care about wall-clock.
 - The service logs under `[h3]`; failures surface with the worker's last output.

tools/h3_common.py

+# CoderAI - OpenAI-compatible API server
+# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+"""MiniMax-H3 rules shared by both ways of running it.
+
+H3 can run either IN the engine process (when the engine's diffusers is >= 0.40)
+or in an isolated venv worker (when it isn't). The arithmetic and the media
+handling are identical either way, and they are checkpoint contracts rather than
+preferences — so they live here, in a module with NO coderai imports and no
+heavy top-level imports, which both sides can load:
+
+  * tools/h3_service.py  imports it as a sibling module inside the isolated venv
+  * codai/api/video.py   loads it by path via codai/api/h3_worker.rules()
+
+Everything here is pure arithmetic or media plumbing; diffusers and torch are
+imported inside the functions that need them.
+"""
+
+import base64
+import os
+import subprocess
+import tempfile
+
+H3_FPS = 24
+H3_FRAMES_CHUNK = 17          # video VAE clip length
+H3_FRAMES_REMAINDER = 5       # decodable counts are 17n + 5
+H3_MIN_SECONDS = 5
+H3_MAX_SECONDS = 15
+H3_CANVAS_MULTIPLE = 32
+
+
+def log(msg: str) -> None:
+    print(f"[h3] {msg}", flush=True)
+
+
+# ── geometry ──────────────────────────────────────────────────────────────────
+
+def snap_frames(num_frames: int) -> int:
+    """Snap up to the next 17n+5 the video VAE can decode, clamped to 5-15 s."""
+    n = max(1, int(num_frames or 0))
+    lo = H3_MIN_SECONDS * H3_FPS       # 120
+    hi = H3_MAX_SECONDS * H3_FPS       # 360
+    n = max(lo, min(hi, n))
+    if n % H3_FRAMES_CHUNK != H3_FRAMES_REMAINDER:
+        n += (H3_FRAMES_REMAINDER - n % H3_FRAMES_CHUNK) % H3_FRAMES_CHUNK
+    while n > hi:                       # snapping may overshoot the 15 s ceiling
+        n -= H3_FRAMES_CHUNK
+    return n
+
+
+def snap_axis(value: int, default: int) -> int:
+    v = int(value or default)
+    v = max(H3_CANVAS_MULTIPLE, v)
+    return v - (v % H3_CANVAS_MULTIPLE)
+
+
+def pick_workflow(body: dict) -> str:
+    """Which transformer partition this request needs.
+
+    Loading with no workflow pulls BOTH partitions (61.7 GB each), so a workflow
+    is always chosen from the inputs.
+    """
+    if body.get("references"):
+        return "ref2va"
+    if body.get("image") or body.get("last_image"):
+        return "fl2va"
+    return "t2va"
+
+
+# ── media helpers ─────────────────────────────────────────────────────────────
+
+def _decode_to_file(data: str, suffix: str) -> str:
+    """Write a base64 / data-URI payload to a temp file and return its path."""
+    if data.startswith("data:"):
+        data = data.split(",", 1)[1]
+    raw = base64.b64decode(data)
+    fd, path = tempfile.mkstemp(suffix=suffix)
+    os.write(fd, raw)
+    os.close(fd)
+    return path
+
+
+def _load_image(data: str):
+    from PIL import Image
+    path = _decode_to_file(data, ".png")
+    try:
+        return Image.open(path).convert("RGB")
+    finally:
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+
+
+def build_references(refs: list, temps: list) -> list:
+    """Turn the request's reference list into H3 reference dataclasses.
+
+    Order matters: it labels them in the prompt presentation and lays them out on
+    the shared rotary clock, so the request order is preserved exactly.
+    """
+    from diffusers.modular_pipelines.minimax_h3 import (
+        MiniMaxH3AudioReference, MiniMaxH3ImageReference, MiniMaxH3VideoReference)
+    kinds = {"image": (MiniMaxH3ImageReference, ".png"),
+             "video": (MiniMaxH3VideoReference, ".mp4"),
+             "audio": (MiniMaxH3AudioReference, ".wav")}
+    out = []
+    for ref in refs or []:
+        kind = (ref.get("type") or "image").lower()
+        if kind not in kinds:
+            raise ValueError(f"unknown reference type '{kind}' (image|video|audio)")
+        cls, suffix = kinds[kind]
+        path = _decode_to_file(ref.get("data") or "", suffix)
+        temps.append(path)
+        out.append(cls.from_file(path))
+    return out
+
+
+def mux(frames, audio, sampling_rate) -> bytes:
+    """Write frames at 24 fps and mux the jointly-generated soundtrack in."""
+    from diffusers.utils import export_to_video
+    tmpdir = tempfile.mkdtemp(prefix="h3_")
+    video_path = os.path.join(tmpdir, "video.mp4")
+    export_to_video(frames, video_path, fps=H3_FPS)
+    out_path = video_path
+
+    if audio is not None and sampling_rate:
+        try:
+            import numpy as np
+            import soundfile as sf
+            wav = audio
+            if hasattr(wav, "detach"):
+                wav = wav.detach().to("cpu").float().numpy()
+            wav = np.asarray(wav)
+            while wav.ndim > 2:            # (1, 2, samples) -> (2, samples)
+                wav = wav[0]
+            if wav.ndim == 2:              # channels-major -> soundfile's (n, ch)
+                wav = wav.T
+            audio_path = os.path.join(tmpdir, "audio.wav")
+            sf.write(audio_path, wav, int(sampling_rate))
+            muxed = os.path.join(tmpdir, "muxed.mp4")
+            proc = subprocess.run(
+                ["ffmpeg", "-y", "-i", video_path, "-i", audio_path,
+                 "-c:v", "copy", "-c:a", "aac", "-b:a", "192k", "-shortest", muxed],
+                stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            if proc.returncode == 0 and os.path.getsize(muxed):
+                out_path = muxed
+            else:
+                log("audio mux failed; returning the silent video: "
+                    + proc.stderr.decode("utf-8", "replace")[-300:])
+        except Exception as exc:
+            log(f"audio mux skipped ({exc}); returning the silent video")
+
+    with open(out_path, "rb") as fh:
+        data = fh.read()
+    import shutil
+    shutil.rmtree(tmpdir, ignore_errors=True)
+    return data

tools/h3_service.py

@@ -61,12 +61,21 @@ import time
 import traceback
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
-H3_FPS = 24
-H3_FRAMES_CHUNK = 17          # video VAE clip length
-H3_FRAMES_REMAINDER = 5       # decodable counts are 17n + 5
-H3_MIN_SECONDS = 5
-H3_MAX_SECONDS = 15
-H3_CANVAS_MULTIPLE = 32
+# The checkpoint contracts (frame/canvas arithmetic, workflow choice, media
+# handling) are shared with the in-engine path — see tools/h3_common.py.
+try:
+    from h3_common import (H3_CANVAS_MULTIPLE, H3_FPS, H3_FRAMES_CHUNK,   # noqa: F401
+                           H3_FRAMES_REMAINDER, H3_MAX_SECONDS, H3_MIN_SECONDS,
+                           build_references, log, mux, pick_workflow,
+                           snap_axis, snap_frames, _decode_to_file, _load_image)
+except ImportError:  # started from elsewhere: load the sibling module by path
+    import importlib.util as _ilu
+    _spec = _ilu.spec_from_file_location(
+        "h3_common", os.path.join(os.path.dirname(os.path.abspath(__file__)), "h3_common.py"))
+    h3_common = _ilu.module_from_spec(_spec)
+    _spec.loader.exec_module(h3_common)
+    from h3_common import *  # noqa: F401,F403  (module registered above)
+    globals().update({k: getattr(h3_common, k) for k in dir(h3_common) if not k.startswith("__")})
 
 _state = {
     "pipe": None,
@@ -80,41 +89,8 @@ _state = {
 }
 
 
-def log(msg: str) -> None:
-    print(f"[h3] {msg}", flush=True)
-
-
-# ── geometry ──────────────────────────────────────────────────────────────────
-
-def snap_frames(num_frames: int) -> int:
-    """Snap up to the next 17n+5 the video VAE can decode, clamped to 5-15 s."""
-    n = max(1, int(num_frames or 0))
-    lo = H3_MIN_SECONDS * H3_FPS       # 120
-    hi = H3_MAX_SECONDS * H3_FPS       # 360
-    n = max(lo, min(hi, n))
-    if n % H3_FRAMES_CHUNK != H3_FRAMES_REMAINDER:
-        n += (H3_FRAMES_REMAINDER - n % H3_FRAMES_CHUNK) % H3_FRAMES_CHUNK
-    while n > hi:                       # snapping may overshoot the 15 s ceiling
-        n -= H3_FRAMES_CHUNK
-    return n
-
-
-def snap_axis(value: int, default: int) -> int:
-    v = int(value or default)
-    v = max(H3_CANVAS_MULTIPLE, v)
-    return v - (v % H3_CANVAS_MULTIPLE)
-
-
 # ── model lifecycle ───────────────────────────────────────────────────────────
 
-def pick_workflow(body: dict) -> str:
-    if body.get("references"):
-        return "ref2va"
-    if body.get("image") or body.get("last_image"):
-        return "fl2va"
-    return "t2va"
-
-
 def _torch_dtype(name: str):
     import torch
     return {"bfloat16": torch.bfloat16, "float16": torch.float16,
@@ -238,96 +214,6 @@ def apply_loras(pipe, loras: list) -> None:
     _state["loras"] = want
 
 
-# ── media helpers ─────────────────────────────────────────────────────────────
-
-def _decode_to_file(data: str, suffix: str) -> str:
-    """Write a base64 / data-URI payload to a temp file and return its path."""
-    if data.startswith("data:"):
-        data = data.split(",", 1)[1]
-    raw = base64.b64decode(data)
-    fd, path = tempfile.mkstemp(suffix=suffix)
-    os.write(fd, raw)
-    os.close(fd)
-    return path
-
-
-def _load_image(data: str):
-    from PIL import Image
-    path = _decode_to_file(data, ".png")
-    try:
-        return Image.open(path).convert("RGB")
-    finally:
-        try:
-            os.unlink(path)
-        except OSError:
-            pass
-
-
-def build_references(refs: list, temps: list) -> list:
-    """Turn the request's reference list into H3 reference dataclasses.
-
-    Order matters: it labels them in the prompt presentation and lays them out on
-    the shared rotary clock, so the request order is preserved exactly.
-    """
-    from diffusers.modular_pipelines.minimax_h3 import (
-        MiniMaxH3AudioReference, MiniMaxH3ImageReference, MiniMaxH3VideoReference)
-    kinds = {"image": (MiniMaxH3ImageReference, ".png"),
-             "video": (MiniMaxH3VideoReference, ".mp4"),
-             "audio": (MiniMaxH3AudioReference, ".wav")}
-    out = []
-    for ref in refs or []:
-        kind = (ref.get("type") or "image").lower()
-        if kind not in kinds:
-            raise ValueError(f"unknown reference type '{kind}' (image|video|audio)")
-        cls, suffix = kinds[kind]
-        path = _decode_to_file(ref.get("data") or "", suffix)
-        temps.append(path)
-        out.append(cls.from_file(path))
-    return out
-
-
-def mux(frames, audio, sampling_rate) -> bytes:
-    """Write frames at 24 fps and mux the jointly-generated soundtrack in."""
-    from diffusers.utils import export_to_video
-    tmpdir = tempfile.mkdtemp(prefix="h3_")
-    video_path = os.path.join(tmpdir, "video.mp4")
-    export_to_video(frames, video_path, fps=H3_FPS)
-    out_path = video_path
-
-    if audio is not None and sampling_rate:
-        try:
-            import numpy as np
-            import soundfile as sf
-            wav = audio
-            if hasattr(wav, "detach"):
-                wav = wav.detach().to("cpu").float().numpy()
-            wav = np.asarray(wav)
-            while wav.ndim > 2:            # (1, 2, samples) -> (2, samples)
-                wav = wav[0]
-            if wav.ndim == 2:              # channels-major -> soundfile's (n, ch)
-                wav = wav.T
-            audio_path = os.path.join(tmpdir, "audio.wav")
-            sf.write(audio_path, wav, int(sampling_rate))
-            muxed = os.path.join(tmpdir, "muxed.mp4")
-            proc = subprocess.run(
-                ["ffmpeg", "-y", "-i", video_path, "-i", audio_path,
-                 "-c:v", "copy", "-c:a", "aac", "-b:a", "192k", "-shortest", muxed],
-                stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-            if proc.returncode == 0 and os.path.getsize(muxed):
-                out_path = muxed
-            else:
-                log("audio mux failed; returning the silent video: "
-                    + proc.stderr.decode("utf-8", "replace")[-300:])
-        except Exception as exc:
-            log(f"audio mux skipped ({exc}); returning the silent video")
-
-    with open(out_path, "rb") as fh:
-        data = fh.read()
-    import shutil
-    shutil.rmtree(tmpdir, ignore_errors=True)
-    return data
-
-
 # ── generation ────────────────────────────────────────────────────────────────
 
 def generate(body: dict) -> dict: