# Copyright (c) 2026 exportAnything. All rights reserved.
# SPDX-License-Identifier: MIT
from __future__ import annotations
import copy
import gc
import hashlib
import json
import logging
import math
import os
import re
import shlex
import subprocess
import threading
import time
import unicodedata
from contextvars import ContextVar
from dataclasses import asdict, dataclass, field
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any
try:
from comfy_execution.graph_utils import ExecutionBlocker
except ModuleNotFoundError: # Standalone repository tests do not add ComfyUI to sys.path.
class ExecutionBlocker: # type: ignore[no-redef]
def __init__(self, message: str | None) -> None:
self.message = message
try:
from .director_cache import (
DIRECTOR_CACHE_KEY_SCHEMA,
DIRECTOR_CACHE_SCHEMA,
cache_path as _director_disk_cache_path,
canonical_sha256 as _canonical_sha256,
checkpoint_identity as _checkpoint_identity,
implementation_identity as _implementation_identity,
load_cache_entry as _load_director_cache_entry,
ordered_tensor_content_hashes as _ordered_tensor_content_hashes,
save_cache_entry as _save_director_cache_entry,
)
from .grounding_guard import (
EVIDENCE_TOKEN_BUDGETS,
GROUNDING_LEDGER_SCHEMA_ID,
GROUNDING_REPORT_SCHEMA_ID,
GroundingGuardConfig,
GroundingGuardDecision,
GroundingLedgerError,
build_asset_registry,
build_grounding_report,
compact_grounding_report,
decide_grounding_guard,
extract_grounding_ledger,
normalize_grounding_guard_config,
persist_grounding_trace,
validate_grounding_ledger,
)
from .grounding_telemetry import build_diffusiongemma_telemetry
from .ltx25_contract import (
LTX25_CAMERA_CAPABILITIES,
LTX25_CONTRACT_SCHEMA,
LTX25_GENERATION_MODES,
LTX25_LONG_HORIZON_MODES,
ltx25_compiler_contract,
ltx25_complexity_budget,
ltx25_long_horizon_plan,
normalize_ltx25_generation_mode,
normalize_ltx25_camera_capability,
normalize_ltx25_long_horizon_mode,
resolve_ltx25_generation_mode,
validate_ltx25_prompt,
)
except ImportError: # Direct-file imports used by the repository smoke tests.
from director_cache import (
DIRECTOR_CACHE_KEY_SCHEMA,
DIRECTOR_CACHE_SCHEMA,
cache_path as _director_disk_cache_path,
canonical_sha256 as _canonical_sha256,
checkpoint_identity as _checkpoint_identity,
implementation_identity as _implementation_identity,
load_cache_entry as _load_director_cache_entry,
ordered_tensor_content_hashes as _ordered_tensor_content_hashes,
save_cache_entry as _save_director_cache_entry,
)
from grounding_guard import (
EVIDENCE_TOKEN_BUDGETS,
GROUNDING_LEDGER_SCHEMA_ID,
GROUNDING_REPORT_SCHEMA_ID,
GroundingGuardConfig,
GroundingGuardDecision,
GroundingLedgerError,
build_asset_registry,
build_grounding_report,
compact_grounding_report,
decide_grounding_guard,
extract_grounding_ledger,
normalize_grounding_guard_config,
persist_grounding_trace,
validate_grounding_ledger,
)
from grounding_telemetry import build_diffusiongemma_telemetry
from ltx25_contract import (
LTX25_CAMERA_CAPABILITIES,
LTX25_CONTRACT_SCHEMA,
LTX25_GENERATION_MODES,
LTX25_LONG_HORIZON_MODES,
ltx25_compiler_contract,
ltx25_complexity_budget,
ltx25_long_horizon_plan,
normalize_ltx25_generation_mode,
normalize_ltx25_camera_capability,
normalize_ltx25_long_horizon_mode,
resolve_ltx25_generation_mode,
validate_ltx25_prompt,
)
CATEGORY = "prompt/diffusiongemma"
COMPATIBILITY_CATEGORY = f"{CATEGORY}/compatibility"
OPTIONAL_CATEGORY = f"{CATEGORY}/optional"
EXPERIMENTAL_MOTION_CATEGORY = f"{CATEGORY}/experimental-motion-planning"
MODEL_TYPE = "DG_MODEL_CONFIG"
MEDIA_TYPE = "DG_MEDIA_CONTEXT"
CONTEXT_TYPE = "DG_CONTEXT"
TARGET_PROFILE_TYPE = "DG_TARGET_PROFILE_CONFIG"
H3_REFERENCE_POLICY_TYPE = "DG_H3_REFERENCE_POLICY_CONFIG"
GROUNDING_GUARD_TYPE = "DG_GROUNDING_GUARD_CONFIG"
GROUNDING_TELEMETRY_SCHEMA_VERSION = "dg-denoising-telemetry/1"
DIRECTOR_RUNTIME_SCHEMA_VERSION = "dg-director-runtime/1"
DIRECTOR_CACHE_MODE_CHOICES = ["reuse", "refresh", "off"]
DIRECTOR_EXECUTION_CONTRACT_REVISION = "h3-reference-policy-v18"
DEFAULT_GGUF_MODEL_PATH = r"Z:\models\LLM\diffusiongemma-26B-A4B-it-GGUF\diffusiongemma-26B-A4B-it-Q4_K_M.gguf"
DEFAULT_GGUF_CLI_PATH = r"C:\tools\llama.cpp-diffusiongemma\build-clang-cuda\bin\llama-diffusion-cli.exe"
DEFAULT_GGUF_EXTRA_ARGS = "-ngl all -cmoe -sm none -mg 0 --diffusion-eb auto --diffusion-kv-cache auto --diffusion-gpu-sampling auto --diffusion-eb-max-steps 24 --diffusion-steps 24 --temp 0.45 -no-cnv --log-verbosity 1"
DEFAULT_CUDA_RUNTIME_BIN = r"C:\tools\cuda-13.3-conda\Library\bin"
DEFAULT_NVFP4_MODEL_DIR = str(Path.cwd() / "models" / "LLM" / "diffusiongemma-26B-A4B-it-NVFP4")
DEFAULT_TRANSFORMERS_MODEL_DIR = str(Path.cwd() / "models" / "LLM" / "diffusiongemma-26B-A4B-it")
DEFAULT_ADVANCED_MODEL_PATH = DEFAULT_NVFP4_MODEL_DIR if Path(DEFAULT_NVFP4_MODEL_DIR).exists() else DEFAULT_TRANSFORMERS_MODEL_DIR
DEFAULT_MODEL_PATH = DEFAULT_GGUF_MODEL_PATH if Path(DEFAULT_GGUF_MODEL_PATH).exists() else DEFAULT_TRANSFORMERS_MODEL_DIR
FINAL_JSON_OPEN = "<|final_json|>"
FINAL_JSON_CLOSE = "<|end_final_json|>"
FINAL_JSON_OPEN_MARKERS = (FINAL_JSON_OPEN, "<|answer|>", "<|final|>")
FINAL_JSON_CLOSE_MARKERS = (FINAL_JSON_CLOSE, "<|/final_json|>", "<|end_final|>", "<|end|>")
MIN_MODELOPT_NVFP4_BRIDGE_MEMORY_GB = 18.0
MEMORY_PRESET_SAFE_20 = "20.0 GB - safe practical"
MEMORY_PRESET_MIN_18 = "18.0 GB - minimum supported"
MEMORY_PRESET_FULL_GPU = "0 GB - full GPU"
MEMORY_PRESET_CHOICES = [MEMORY_PRESET_SAFE_20, MEMORY_PRESET_MIN_18, MEMORY_PRESET_FULL_GPU]
RESOLUTION_SELECTOR_ASPECT_LABELS = {
"1:1": "1:1 (Square)",
"2:3": "2:3 (Portrait Photo)",
"3:2": "3:2 (Photo)",
"3:4": "3:4 (Portrait Standard)",
"4:3": "4:3 (Standard)",
"9:16": "9:16 (Portrait Widescreen)",
"16:9": "16:9 (Widescreen)",
"21:9": "21:9 (Ultrawide)",
}
RESOLUTION_SELECTOR_AUTO = "Auto (target/source)"
RESOLUTION_SELECTOR_ASPECT_CHOICES = [
RESOLUTION_SELECTOR_AUTO,
*RESOLUTION_SELECTOR_ASPECT_LABELS.values(),
]
RESOLUTION_SELECTOR_RATIOS = {
"1:1": (1, 1),
"2:3": (2, 3),
"3:2": (3, 2),
"3:4": (3, 4),
"4:3": (4, 3),
"9:16": (9, 16),
"16:9": (16, 9),
"21:9": (21, 9),
}
REFERENCE_PREP_ASPECT_CHOICES = ["source", "1:1", "16:9", "9:16", "4:3", "3:4", "3:2", "2:3", "21:9"]
MEDIA_SYNTHESIS_MODE_CHOICES = ["video_recreation", "image_identity_video_control"]
DEFAULT_MASTER_PROMPT = """You are a local ComfyUI prompt architect. Treat the user's text as a content brief: it only needs to say what the video or image should be about. Do not require the user to ask for prompt-writing boilerplate such as detailed subject, background, lighting, camera movement, pacing, composition, or style; infer useful generator-ready detail when it is missing, while preserving every explicit user constraint. Model-specific target rules below override this general expansion rule. Preserve an explicitly requested or evidenced static hold. A conditioned still frame fixes its camera geometry at that instant; it does not by itself prove that the camera remains locked afterward.
Use the user's text, any backend-consumable pixels, and any supplied visual_description to produce final, practical prompts for generation models. If the active backend is metadata-only for image/video, do not pretend you saw pixels; use only the user's text, the visual_description, and factual media metadata. Preserve requested subjects, exact text, composition, motion, style, camera behavior, visual capture language, audio mode, audio guidance, and constraints. In image_identity_video_control workflows, treat a request to replace the video subject with the image subject as a strict identity-and-appearance transfer: face, hair, body type, wardrobe, accessories, styling, and distinguishing traits come from the reference image, while the video contributes pose, action, blocking, depth/canny/edge structure, camera choreography, timing, composition, and scene geometry. Do not copy the control-video subject's wardrobe, hair, face, body type, accessories, or styling into the final prompt unless the user explicitly asks for those video-subject traits. For LTX source-video work, explicitly observe camera choreography before writing the final prompt: first-frame and ending framing, shot size, camera height/angle, camera support or stability, camera path, zoom/dolly/truck/boom/pan/tilt/roll/orbit/rotation, subject-camera relationship, focus behavior, depth/parallax, and visible lens/capture changes. Transcribe observed camera work precisely instead of reducing it to generic phrases like camera movement, zoom out, or pan left. Visual capture language includes exposure level, overexposed or blown highlights, underexposure, flash/direct light, bloom, halation, lens blur, bokeh, depth of field, focal length impression, perspective compression, lens distortion, motion blur, grain/noise, white balance, contrast, dynamic range, and camera angle. Do not describe invisible intent. Faithful mode preserves explicitly requested camera energy. For LTX targets only, that preservation is bounded by the LTX Target Profile's Camera Capability: Stable / base model simplifies unsupported risky camera movement, while Advanced / controlled camera preserves the prior fast or compound choreography policy. Other targets do not inherit this LTX-only capability gate and must follow their own model-specific camera contract. Every prompt field must contain the finished prompt text or finished prompt JSON, never instructions about how to write a prompt. Return only valid JSON with ltx_prompt, ideogram_prompt, minimax_h3_prompt, negative_prompt, scene_segments, and metadata."""
MINIMAX_H3_MASTER_PROMPT = """You are a local ComfyUI MiniMax H3 prompt architect. Treat the user's text as a content brief: it only needs to say what the video should be about. Infer useful generator-ready subject, environment, lighting, camera, motion, pacing, composition, style, and sound detail when it is missing while preserving every explicit user constraint. Preserve an explicitly requested or evidenced static hold. A conditioned still frame fixes its camera geometry at that instant; it does not by itself prove that the camera remains locked afterward.
Use the user's text, any backend-consumable pixels, and any supplied visual_description to produce one final, practical MiniMax H3 prompt. If the active backend is metadata-only, do not pretend you saw pixels; use only the user's text, the visual_description, and factual media metadata. Preserve requested subjects, exact text, composition, motion, style, camera behavior, visual capture language, audio guidance, and constraints. Visual capture language includes exposure level, blown highlights, underexposure, flash or direct light, bloom, halation, lens blur, bokeh, depth of field, focal length impression, perspective compression, lens distortion, motion blur, grain or noise, white balance, contrast, dynamic range, and camera angle. Do not describe invisible intent. The finished prompt must never contain instructions about how to write a prompt. Return only the JSON object requested by the model-facing contract below."""
@dataclass
class RuntimeConfig:
model_path: str
backend: str
dtype: str
quantization: str
local_files_only: bool
unload_policy: str
max_memory_gb: float
cli_path: str = ""
extra_args: str = ""
temperature: float = 0.45
fallback_backend: str = "template"
status: dict[str, Any] = field(default_factory=dict)
@dataclass
class MediaContext:
images: Any = None
source: str = "none"
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass
class GemmaContext:
user_prompt: str = ""
images: Any = None
source: str = "none"
media_metadata: dict[str, Any] = field(default_factory=dict)
visual_description: str = ""
warnings: list[str] = field(default_factory=list)
@dataclass
class TargetProfileConfig:
target_profile: str = "ltx"
audio_mode: str = "auto_scene_audio"
audio_guidance: str = ""
target_duration_seconds: float = 0.0
ltx_style: str = ""
ideogram_aspect_ratio: str = "1:1"
ideogram_render_style: str = ""
ideogram_exact_text: str = ""
ideogram_json_output: bool = True
negative_prompt_mode: str = "auto"
negative_prompt_guidance: str = ""
minimax_h3_mode: str = "t2va"
minimax_h3_shot_count: str = "auto"
minimax_h3_dialogue_mode: str = "auto"
minimax_h3_dialogue_line_count: int = 2
minimax_h3_dialogue_guidance: str = ""
ltx_generation_mode: str = "auto"
ltx_long_horizon_mode: str = "off"
ltx_camera_capability: str = "stable"
@dataclass
class H3ReferencePolicyConfig:
"""Host-owned rules for turning connected Ref2VA media into a manifest."""
layout: str = "auto"
custom_manifest: str = ""
expected_subject_count: int = 0
@dataclass
class BackendRunOptions:
"""Per-call controls used only by the Grounding Guard path.
A ``None`` options value is deliberately meaningful: it selects the
historical backend call without changing its processor or RNG behavior.
"""
sampling_profile: str = "checkpoint_defaults"
thinking_mode: str = "auto"
seed: int = 0
enable_telemetry: bool = True
require_transport: bool = False
fail_closed_telemetry: bool = False
visual_first: bool = True
asset_registry: dict[str, Any] = field(default_factory=dict)
call_budget: "BackendCallBudget | None" = None
stage: str = "backend"
@dataclass
class BackendRunResult:
decoded_text: str
native_decoded_text: str = ""
stage: str = "backend"
transport: dict[str, Any] = field(default_factory=dict)
telemetry: dict[str, Any] = field(default_factory=dict)
effective_sampling: dict[str, Any] = field(default_factory=dict)
timing: dict[str, Any] = field(default_factory=dict)
forward_counts: dict[str, Any] = field(default_factory=dict)
input_tokens: int = 0
generated_tokens: int = 0
generated_characters: int = 0
class VisualTransportError(RuntimeError):
def __init__(self, message: str, transport: dict[str, Any] | None = None) -> None:
super().__init__(message)
self.transport = copy.deepcopy(transport or {})
@dataclass
class BackendCallBudget:
max_calls: int = 4
attempted_calls: int = 0
def claim(self) -> int:
if self.attempted_calls >= self.max_calls:
raise RuntimeError(f"Grounding Guard exceeded its hard {self.max_calls}-call ceiling.")
self.attempted_calls += 1
return self.attempted_calls
_MODEL_CACHE: dict[str, tuple[Any, Any]] = {}
_NVFP4_SMOKE_CACHE: dict[str, dict[str, Any]] = {}
_CURRENT_DIRECTOR_RUNTIME: ContextVar[dict[str, Any] | None] = ContextVar(
"diffusiongemma_director_runtime",
default=None,
)
def _new_director_runtime_metrics(cache_mode: str) -> dict[str, Any]:
return {
"schema": DIRECTOR_RUNTIME_SCHEMA_VERSION,
"model_load_seconds": 0.0,
"nvfp4_bridge_seconds": 0.0,
"media_preprocess_seconds": 0.0,
# DiffusionGemma performs multimodal encoding inside generate(); the
# current Transformers surface does not expose a safe boundary for it.
"multimodal_prefill_seconds": 0.0,
"multimodal_prefill_is_separately_measured": False,
"generation_seconds": 0.0,
"grounding_validation_seconds": 0.0,
"grounding_attempt_count": 0,
"generated_token_count": 0,
"generated_character_count": 0,
"model_unload_seconds": 0.0,
"total_director_seconds": 0.0,
"peak_vram_mb": 0,
"model_call_count": 0,
"target_repair_attempt_count": 0,
"dialogue_patch_attempt_count": 0,
"retry_reason_count": 0,
"grounding_first_pass_accepted": False,
"cache_mode": str(cache_mode or "reuse"),
"cache_hit": False,
"cache_lookup_seconds": 0.0,
"cache_write_seconds": 0.0,
"runtime_model_cache_hit_count": 0,
"runtime_model_cold_load_count": 0,
"calls": [],
"notes": {
"nvfp4_bridge_seconds": "ModelOpt expert-bank packing/loading; this is a subset of model_load_seconds.",
"media_preprocess_seconds": "Tensor-to-PIL conversion, chat-template processing, and input transfer inside the Director; upstream Media Sampler decode is not included.",
"multimodal_prefill_seconds": "Included in generation_seconds until Transformers exposes a safe separate boundary.",
"peak_vram_mb": "Process-wide CUDA peak allocated memory measured only for cache misses.",
},
}
def _director_metric_add(name: str, value: float | int) -> None:
metrics = _CURRENT_DIRECTOR_RUNTIME.get()
if not isinstance(metrics, dict):
return
current = metrics.get(name, 0)
if isinstance(current, int) and isinstance(value, int):
metrics[name] = current + value
else:
metrics[name] = _safe_float(current, 0.0) + _safe_float(value, 0.0)
def _director_metric_set(name: str, value: Any) -> None:
metrics = _CURRENT_DIRECTOR_RUNTIME.get()
if isinstance(metrics, dict):
metrics[name] = value
def _director_metric_append_call(value: dict[str, Any]) -> None:
metrics = _CURRENT_DIRECTOR_RUNTIME.get()
if not isinstance(metrics, dict):
return
calls = metrics.setdefault("calls", [])
if isinstance(calls, list):
calls.append(copy.deepcopy(value))
def _reset_director_peak_vram() -> bool:
try:
import torch
if not torch.cuda.is_available():
return False
torch.cuda.reset_peak_memory_stats()
return True
except Exception:
return False
def _director_peak_vram_mb() -> int:
try:
import torch
if not torch.cuda.is_available():
return 0
return int(round(float(torch.cuda.max_memory_allocated()) / (1024.0 * 1024.0)))
except Exception:
return 0
def _rounded_director_runtime(metrics: dict[str, Any]) -> dict[str, Any]:
result = copy.deepcopy(metrics)
for key in (
"model_load_seconds",
"nvfp4_bridge_seconds",
"media_preprocess_seconds",
"multimodal_prefill_seconds",
"generation_seconds",
"grounding_validation_seconds",
"model_unload_seconds",
"total_director_seconds",
"cache_lookup_seconds",
"cache_write_seconds",
):
result[key] = round(max(0.0, _safe_float(result.get(key), 0.0)), 6)
for key in (
"grounding_attempt_count",
"generated_token_count",
"generated_character_count",
"peak_vram_mb",
"model_call_count",
"target_repair_attempt_count",
"dialogue_patch_attempt_count",
"retry_reason_count",
"runtime_model_cache_hit_count",
"runtime_model_cold_load_count",
):
result[key] = int(max(0.0, _safe_float(result.get(key), 0.0)))
return result
def _dg_log(message: str, *args: Any) -> None:
logging.info("[DiffusionGemma] " + message, *args)
def _progress_enabled() -> bool:
return os.environ.get("DG_PROGRESS", "1").strip().lower() not in {"0", "false", "off", "no"}
def _release_transformers_runtime() -> None:
_MODEL_CACHE.clear()
gc.collect()
try:
import torch
if torch.cuda.is_available():
torch.cuda.empty_cache()
try:
torch.cuda.ipc_collect()
except Exception:
pass
except Exception:
pass
def _json_dumps(data: Any) -> str:
return json.dumps(data, indent=2, sort_keys=True, ensure_ascii=True)
def _prompt_json_dumps(data: Any) -> str:
return json.dumps(data, indent=2, ensure_ascii=False)
def _ideogram_json_dumps(data: Any) -> str:
return json.dumps(data, separators=(",", ":"), ensure_ascii=False)
def _safe_float(value: Any, default: float = 0.0) -> float:
try:
return float(value)
except Exception:
return default
def _memory_preset_to_gb(value: Any) -> float:
if isinstance(value, (int, float)):
return float(value)
text = str(value or "").strip()
lowered = text.lower()
if "full" in lowered:
return 0.0
if "minimum" in lowered:
return MIN_MODELOPT_NVFP4_BRIDGE_MEMORY_GB
if "safe" in lowered:
return 20.0
match = re.search(r"[-+]?\d+(?:\.\d+)?", text)
if match:
return _safe_float(match.group(0), 20.0)
return 20.0
def _run_blocking_with_progress(
label: str,
fn: Any,
*,
node_id: str | None = None,
estimated_seconds: float | None = None,
) -> Any:
if not _progress_enabled():
return fn()
estimate = max(5.0, _safe_float(os.environ.get("DG_PROGRESS_ESTIMATE_SECONDS", ""), estimated_seconds or 120.0))
estimate_units = int(math.ceil(estimate))
interval = max(0.25, _safe_float(os.environ.get("DG_PROGRESS_INTERVAL_SECONDS", ""), 1.0))
result_box: dict[str, Any] = {}
def worker() -> None:
try:
result_box["value"] = fn()
except BaseException as exc:
result_box["error"] = exc
progress_bar = None
tqdm_bar = None
try:
from comfy.utils import ProgressBar
progress_bar = ProgressBar(estimate_units, node_id=node_id)
progress_bar.update_absolute(0, estimate_units)
except Exception:
progress_bar = None
try:
from tqdm.auto import tqdm
tqdm_bar = tqdm(
total=estimate_units,
desc=label,
unit="s",
leave=True,
bar_format="{l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}]",
)
except Exception:
tqdm_bar = None
thread = threading.Thread(target=worker, name=f"{label} worker", daemon=True)
start = time.perf_counter()
last_reported = 0
displayed_total = estimate_units
thread.start()
try:
while thread.is_alive():
elapsed = time.perf_counter() - start
displayed_total = max(displayed_total, int(math.ceil(elapsed + interval)))
current = min(int(elapsed), displayed_total)
if progress_bar is not None:
progress_bar.update_absolute(current, displayed_total)
if tqdm_bar is not None:
if displayed_total != tqdm_bar.total:
tqdm_bar.total = displayed_total
tqdm_bar.refresh()
delta = max(0, current - last_reported)
if delta:
tqdm_bar.update(delta)
last_reported += delta
time.sleep(interval)
thread.join()
elapsed = time.perf_counter() - start
total = max(1, last_reported, int(math.ceil(elapsed)))
if progress_bar is not None:
progress_bar.update_absolute(total, total)
if tqdm_bar is not None:
if total != tqdm_bar.total:
tqdm_bar.total = total
remaining = max(0.0, total - tqdm_bar.n)
if remaining:
tqdm_bar.update(remaining)
finally:
if tqdm_bar is not None:
tqdm_bar.close()
if "error" in result_box:
raise result_box["error"]
return result_box.get("value")
def _normalize_aspect_ratio_key(aspect_ratio: Any) -> str:
text = str(aspect_ratio or "").strip()
match = re.search(r"\b(\d+)\s*:\s*(\d+)\b", text)
if not match:
return "1:1"
key = f"{int(match.group(1))}:{int(match.group(2))}"
return key if key in RESOLUTION_SELECTOR_RATIOS else "1:1"
def _resolution_selector_preset(aspect_ratio: Any) -> str:
return RESOLUTION_SELECTOR_ASPECT_LABELS[_normalize_aspect_ratio_key(aspect_ratio)]
def _known_aspect_ratio_label_from_dimensions(width: int, height: int) -> str:
src_w = max(1, int(width))
src_h = max(1, int(height))
source_ratio = src_w / float(src_h)
best_key = ""
best_delta = float("inf")
for key, (w_ratio, h_ratio) in RESOLUTION_SELECTOR_RATIOS.items():
delta = abs(source_ratio - (w_ratio / float(h_ratio)))
if delta < best_delta:
best_delta = delta
best_key = key
if best_key and best_delta <= 0.02:
return RESOLUTION_SELECTOR_ASPECT_LABELS[best_key]
return f"Custom / source ({src_w}x{src_h})"
def _resolution_selector_dimensions(aspect_ratio: Any, megapixels: Any = 2.0, multiple: Any = 8) -> tuple[int, int]:
key = _normalize_aspect_ratio_key(aspect_ratio)
w_ratio, h_ratio = RESOLUTION_SELECTOR_RATIOS[key]
mp = max(0.1, min(16.0, _safe_float(megapixels, 2.0)))
mult = int(max(8, min(128, _safe_float(multiple, 8))))
mult = max(1, int(round(mult / 4)) * 4)
total_pixels = mp * 1024 * 1024
scale = math.sqrt(total_pixels / (w_ratio * h_ratio))
width = max(mult, round(w_ratio * scale / mult) * mult)
height = max(mult, round(h_ratio * scale / mult) * mult)
return int(width), int(height)
def _nearest_resolution_aspect_ratio_key(width: Any, height: Any) -> str:
src_w = max(1, int(_safe_float(width, 1)))
src_h = max(1, int(_safe_float(height, 1)))
source_ratio = src_w / float(src_h)
return min(
RESOLUTION_SELECTOR_RATIOS,
key=lambda key: abs(
source_ratio
- (
RESOLUTION_SELECTOR_RATIOS[key][0]
/ float(RESOLUTION_SELECTOR_RATIOS[key][1])
)
),
)
def _splitter_resolution_aspect_ratio(
requested: Any,
context: "GemmaContext",
target: dict[str, Any],
packet_aspect_ratio: Any,
) -> tuple[str, str]:
selection = str(requested or RESOLUTION_SELECTOR_AUTO).strip()
profile = _normalize_target_profile(str(target.get("target_profile", "ltx")))
if profile == "ideogram4":
return _normalize_aspect_ratio_key(packet_aspect_ratio), "ideogram_target_profile"
auto_aliases = {
"",
"auto",
"auto (recommended)",
RESOLUTION_SELECTOR_AUTO.casefold(),
}
if selection.casefold() not in auto_aliases:
match = re.search(r"\b(\d+)\s*:\s*(\d+)\b", selection)
if match:
requested_key = f"{int(match.group(1))}:{int(match.group(2))}"
if requested_key in RESOLUTION_SELECTOR_RATIOS:
return requested_key, "splitter_widget"
if profile == "ltx":
metadata = context.media_metadata if isinstance(context, GemmaContext) else {}
for width_key, height_key, source in (
("ltx_first_frame_width", "ltx_first_frame_height", "ltx_first_frame"),
("width", "height", "media_dimensions"),
):
width = int(_safe_float(metadata.get(width_key), 0))
height = int(_safe_float(metadata.get(height_key), 0))
if width > 0 and height > 0:
return _nearest_resolution_aspect_ratio_key(width, height), source
return "16:9", "ltx_default"
return _normalize_aspect_ratio_key(packet_aspect_ratio), "legacy_target_profile"
def _round_dimension_to_multiple(value: float, multiple: int) -> int:
mult = int(max(1, multiple))
return max(mult, int(round(float(value) / mult) * mult))
def _dimensions_for_long_edge(width: int, height: int, long_edge: int, multiple: int = 8) -> tuple[int, int]:
src_w = max(1, int(width))
src_h = max(1, int(height))
edge = max(64, int(long_edge))
scale = edge / float(max(src_w, src_h))
return (
_round_dimension_to_multiple(src_w * scale, multiple),
_round_dimension_to_multiple(src_h * scale, multiple),
)
def _aspect_dimensions_for_long_edge(aspect_ratio: Any, long_edge: int, multiple: int = 8) -> tuple[int, int]:
key = _normalize_aspect_ratio_key(aspect_ratio)
w_ratio, h_ratio = RESOLUTION_SELECTOR_RATIOS[key]
edge = max(64, int(long_edge))
if w_ratio >= h_ratio:
width = edge
height = max(1, int(round(edge * (h_ratio / float(w_ratio)))))
else:
height = edge
width = max(1, int(round(edge * (w_ratio / float(h_ratio)))))
return (_round_dimension_to_multiple(width, multiple), _round_dimension_to_multiple(height, multiple))
def _center_crop_dimensions_for_aspect(width: int, height: int, aspect_ratio: Any) -> tuple[int, int]:
key = _normalize_aspect_ratio_key(aspect_ratio)
target_w, target_h = RESOLUTION_SELECTOR_RATIOS[key]
src_w = max(1, int(width))
src_h = max(1, int(height))
src_ratio = src_w / float(src_h)
dst_ratio = target_w / float(target_h)
if abs(src_ratio - dst_ratio) < 1e-6:
return (src_w, src_h)
if src_ratio > dst_ratio:
crop_w = max(1, int(round(src_h * dst_ratio)))
return (min(crop_w, src_w), src_h)
crop_h = max(1, int(round(src_w / dst_ratio)))
return (src_w, min(crop_h, src_h))
def _center_crop_image_batch(image: Any, crop_width: int, crop_height: int) -> Any:
try:
import torch
tensor = image if isinstance(image, torch.Tensor) else torch.as_tensor(image)
if tensor.ndim == 3:
tensor = tensor.unsqueeze(0)
_, src_h, src_w, _ = tensor.shape
crop_w = max(1, min(int(crop_width), int(src_w)))
crop_h = max(1, min(int(crop_height), int(src_h)))
x0 = max(0, (int(src_w) - crop_w) // 2)
y0 = max(0, (int(src_h) - crop_h) // 2)
return tensor[:, y0 : y0 + crop_h, x0 : x0 + crop_w, :]
except Exception:
return image
def _resize_image_batch(image: Any, width: int, height: int) -> Any:
import torch
tensor = image if isinstance(image, torch.Tensor) else torch.as_tensor(image)
if tensor.ndim == 3:
tensor = tensor.unsqueeze(0)
original_dtype = tensor.dtype
original_device = tensor.device
work = tensor
if not torch.is_floating_point(work):
work = work.float()
nchw = work.permute(0, 3, 1, 2).contiguous()
try:
resized = torch.nn.functional.interpolate(
nchw,
size=(max(1, int(height)), max(1, int(width))),
mode="bilinear",
align_corners=False,
antialias=True,
)
except TypeError:
resized = torch.nn.functional.interpolate(
nchw,
size=(max(1, int(height)), max(1, int(width))),
mode="bilinear",
align_corners=False,
)
result = resized.permute(0, 2, 3, 1).contiguous()
if torch.is_floating_point(tensor) and result.dtype != original_dtype:
result = result.to(dtype=original_dtype)
return result.to(device=original_device)
def _h3_reference_dimensions_for_pixel_budget(
width: int,
height: int,
pixel_budget: float,
multiple: int = 32,
) -> tuple[int, int]:
"""Return downscale-only, aspect-preserving dimensions within a pixel budget."""
src_w = max(1, int(width))
src_h = max(1, int(height))
mult = max(1, int(multiple))
minimum_pixels = mult * mult
budget = float(pixel_budget)
if budget < minimum_pixels:
raise ValueError(
f"MiniMax H3 reference pixel budget must allow at least {mult}x{mult} pixels; got {budget:.0f}."
)
scale = min(1.0, math.sqrt(budget / float(src_w * src_h)))
target_w = max(mult, int(math.floor((src_w * scale) / mult)) * mult)
target_h = max(mult, int(math.floor((src_h * scale) / mult)) * mult)
target_w = min(target_w, max(mult, (src_w // mult) * mult))
target_h = min(target_h, max(mult, (src_h // mult) * mult))
if target_w > src_w or target_h > src_h:
raise ValueError(
f"MiniMax H3 references must be at least {mult} pixels on each axis; got {src_w}x{src_h}."
)
while target_w * target_h > budget and (target_w > mult or target_h > mult):
width_scale = target_w / float(src_w)
height_scale = target_h / float(src_h)
if target_w > mult and (target_h == mult or width_scale >= height_scale):
target_w -= mult
else:
target_h -= mult
if target_w * target_h > budget:
raise ValueError(
f"MiniMax H3 reference pixel budget {budget:.0f} cannot fit a {mult}x{mult} aligned image."
)
return (int(target_w), int(target_h))
def _h3_native_match_reference_dimensions(
width: int,
height: int,
generation_width: int,
generation_height: int,
multiple: int = 32,
) -> tuple[int, int]:
"""Mirror MiniMaxH3ReferenceToVideo's current ``match`` sizing for telemetry."""
src_w = max(1, int(width))
src_h = max(1, int(height))
canvas_pixels = max(1, int(generation_width)) * max(1, int(generation_height))
scale = min(1.0, math.sqrt(canvas_pixels / float(src_w * src_h)))
mult = max(1, int(multiple))
return (
max(mult, int(round((src_w * scale) / mult)) * mult),
max(mult, int(round((src_h * scale) / mult)) * mult),
)
def _h3_shared_reference_pixel_budgets(
source_pixels_1: int,
source_pixels_2: int,
total_pixel_budget: float,
reference_1_share: float,
) -> tuple[float, float]:
"""Split one budget between two refs and reassign unused source capacity."""
total = max(0.0, float(total_pixel_budget))
share = max(0.0, min(1.0, float(reference_1_share)))
budget_1 = total * share
budget_2 = total - budget_1
if source_pixels_1 < budget_1:
budget_2 += budget_1 - float(source_pixels_1)
budget_1 = float(source_pixels_1)
if source_pixels_2 < budget_2:
budget_1 += budget_2 - float(source_pixels_2)
budget_2 = float(source_pixels_2)
return (budget_1, budget_2)
def _read_json_file(path: Path) -> dict[str, Any]:
try:
data = json.loads(path.read_text(encoding="utf-8"))
return data if isinstance(data, dict) else {}
except Exception:
return {}
def _pad_nvfp4_block_scale_rows(block_scale: Any, output_rows: int) -> Any:
import torch
rounded_rows = ((int(output_rows) + 127) // 128) * 128
if int(block_scale.shape[0]) >= rounded_rows:
return block_scale
pad_shape = (rounded_rows - int(block_scale.shape[0]), int(block_scale.shape[1]))
padding = torch.zeros(pad_shape, dtype=block_scale.dtype, device=block_scale.device)
return torch.cat([block_scale, padding], dim=0)
def _modelopt_to_comfy_fp4_packing(qdata: Any) -> Any:
return ((qdata & 0x0F) << 4) | ((qdata & 0xF0) >> 4)
def _modelopt_to_comfy_block_scale(block_scale: Any) -> Any:
from comfy_kitchen.float_utils import to_blocked
return to_blocked(block_scale, flatten=False)
def _safetensors_index(model_path: str) -> dict[str, str]:
index_path = Path(model_path) / "model.safetensors.index.json"
index_data = _read_json_file(index_path)
weight_map = index_data.get("weight_map")
return weight_map if isinstance(weight_map, dict) else {}
def _nvfp4_bridge_smoke(model_path: str) -> dict[str, Any]:
cache_key = str(Path(model_path))
cached = _NVFP4_SMOKE_CACHE.get(cache_key)
if cached is not None:
return cached.copy()
result: dict[str, Any] = {"passed": False, "error": "", "projection": ""}
try:
import torch
from safetensors import safe_open
from comfy.quant_ops import QuantizedTensor, get_layout_class
if not torch.cuda.is_available():
raise RuntimeError("CUDA is not available.")
root = Path(model_path)
weight_map = _safetensors_index(model_path)
prefix = "model.decoder.layers.0.experts.0.gate_proj"
required = [f"{prefix}.{name}" for name in ("weight", "weight_scale", "weight_scale_2", "input_scale")]
if not all(key in weight_map for key in required):
raise RuntimeError("Expected layer-0 expert gate_proj NVFP4 tensors were not found.")
handles: dict[str, Any] = {}
def get_tensor(key: str) -> Any:
shard = weight_map[key]
handle = handles.get(shard)
if handle is None:
handle = safe_open(str(root / shard), framework="pt", device="cpu")
handles[shard] = handle
return handle.get_tensor(key)
try:
qdata = _modelopt_to_comfy_fp4_packing(get_tensor(f"{prefix}.weight")).to("cuda", non_blocking=True)
block_scale = _modelopt_to_comfy_block_scale(get_tensor(f"{prefix}.weight_scale")).to("cuda", non_blocking=True)
tensor_scale = get_tensor(f"{prefix}.weight_scale_2").to("cuda", non_blocking=True)
input_scale = get_tensor(f"{prefix}.input_scale").to("cuda", non_blocking=True)
finally:
handles.clear()
out_features = int(qdata.shape[0])
in_features = int(qdata.shape[1]) * 2
layout_cls = get_layout_class("TensorCoreNVFP4Layout")
weight_params = layout_cls.Params(
scale=tensor_scale,
block_scale=block_scale,
orig_dtype=torch.bfloat16,
orig_shape=(out_features, in_features),
)
weight = QuantizedTensor(qdata.to(torch.uint8), "TensorCoreNVFP4Layout", weight_params)
x = torch.zeros((2, in_features), dtype=torch.bfloat16, device="cuda")
qx = QuantizedTensor.from_float(x, "TensorCoreNVFP4Layout", scale=input_scale)
y = torch.nn.functional.linear(qx, weight, None)
torch.cuda.synchronize()
if not bool(torch.isfinite(y.float()).all()):
raise RuntimeError("NVFP4 bridge smoke produced non-finite output.")
result.update(
{
"passed": True,
"projection": prefix,
"output_shape": list(y.shape),
"output_dtype": str(y.dtype).replace("torch.", ""),
}
)
except Exception as exc:
result["error"] = f"{type(exc).__name__}: {exc}"
finally:
gc.collect()
try:
import torch
if torch.cuda.is_available():
torch.cuda.empty_cache()
except Exception:
pass
_NVFP4_SMOKE_CACHE[cache_key] = result.copy()
return result
def _model_path_info(model_path: str) -> dict[str, Any]:
path = Path(model_path or "")
info: dict[str, Any] = {
"path_kind": "missing",
"is_gguf_file": False,
"is_hf_repo": False,
"model_type": "",
"architecture": "",
"quant_algo": "",
"quant_method": "",
"quant_producer": "",
"quant_producer_version": "",
"transformers_version_required": "",
"has_generation_config": False,
"has_modelopt_state": False,
"runtime_hint": "",
}
if not path.exists():
return info
if path.is_file():
info["path_kind"] = "gguf_file" if path.suffix.lower() == ".gguf" else "file"
info["is_gguf_file"] = path.suffix.lower() == ".gguf"
if info["is_gguf_file"]:
info["runtime_hint"] = "llama-diffusion-cli GGUF"
return info
if not path.is_dir():
info["path_kind"] = "unknown"
return info
config = _read_json_file(path / "config.json")
hf_quant_config = _read_json_file(path / "hf_quant_config.json")
quant_config = config.get("quantization_config") if isinstance(config.get("quantization_config"), dict) else {}
if not quant_config:
quant_config = hf_quant_config.get("quantization") if isinstance(hf_quant_config.get("quantization"), dict) else {}
producer = quant_config.get("producer") if isinstance(quant_config.get("producer"), dict) else {}
if not producer:
producer = hf_quant_config.get("producer") if isinstance(hf_quant_config.get("producer"), dict) else {}
architectures = config.get("architectures") if isinstance(config.get("architectures"), list) else []
info.update(
{
"is_hf_repo": bool(config),
"model_type": str(config.get("model_type", "")),
"architecture": str(architectures[0]) if architectures else "",
"quant_algo": str(quant_config.get("quant_algo", "")),
"quant_method": str(quant_config.get("quant_method", config.get("quantization_config", {}).get("quant_method", "")) if isinstance(config.get("quantization_config"), dict) else ""),
"quant_producer": str(producer.get("name", "")),
"quant_producer_version": str(producer.get("version", "")),
"transformers_version_required": str(config.get("transformers_version", "")),
"has_generation_config": (path / "generation_config.json").exists(),
"has_modelopt_state": (path / "modelopt_state.pth").exists(),
}
)
if info["quant_algo"].upper() == "NVFP4":
info["path_kind"] = "nvfp4_hf_repo"
info["runtime_hint"] = "NVIDIA ModelOpt NVFP4 checkpoint"
elif config:
info["path_kind"] = "hf_repo"
info["runtime_hint"] = "Transformers whole-repo model"
else:
info["path_kind"] = "directory"
return info
def _runtime_status(config: RuntimeConfig) -> dict[str, Any]:
model_info = _model_path_info(config.model_path)
status: dict[str, Any] = {
"backend": config.backend,
"model_path": config.model_path,
"model_path_exists": Path(config.model_path).exists(),
**model_info,
"local_files_only": config.local_files_only,
"max_memory_gb": config.max_memory_gb,
"temperature": config.temperature,
"temperature_semantics": (
"GGUF compatibility scalar; in-process DiffusionGemma Grounding Guard calls use the selected native "
"0.8-to-0.4 denoising sampling profile."
),
"transformers_importable": False,
"diffusion_gemma_module": False,
"diffusion_gemma_class": False,
"auto_processor": False,
"vllm_importable": False,
"modelopt_importable": False,
"comfy_kitchen_importable": False,
"fp_quant_importable": False,
"qutlass_importable": False,
"tensorrt_llm_importable": False,
"torch_cuda_available": False,
"torch_cuda_capability": "",
"transformers_version": "",
"processor_class": "",
"modelopt_nvfp4_supported": False,
"comfy_nvfp4_bridge_supported": False,
"nvfp4_bridge_min_memory_gb": MIN_MODELOPT_NVFP4_BRIDGE_MEMORY_GB,
"nvfp4_bridge_memory_guard_passed": True,
"nvfp4_bridge_smoke": {},
"ready": False,
"supports_pixels": False,
"supports_video": False,
"supports_thinking": False,
"media_policy": "unknown",
"notes": [],
}
try:
import importlib.util
status["transformers_importable"] = importlib.util.find_spec("transformers") is not None
status["diffusion_gemma_module"] = importlib.util.find_spec("transformers.models.diffusion_gemma") is not None
status["vllm_importable"] = importlib.util.find_spec("vllm") is not None
status["modelopt_importable"] = bool(
importlib.util.find_spec("modelopt") or importlib.util.find_spec("nvidia_modelopt")
)
status["comfy_kitchen_importable"] = importlib.util.find_spec("comfy_kitchen") is not None
status["fp_quant_importable"] = importlib.util.find_spec("fp_quant") is not None
status["qutlass_importable"] = importlib.util.find_spec("qutlass") is not None
status["tensorrt_llm_importable"] = importlib.util.find_spec("tensorrt_llm") is not None
try:
import transformers
status["transformers_version"] = str(getattr(transformers, "__version__", ""))
except Exception:
pass
try:
import torch
status["torch_cuda_available"] = bool(torch.cuda.is_available())
if torch.cuda.is_available():
status["torch_cuda_capability"] = ".".join(str(part) for part in torch.cuda.get_device_capability())
except Exception as exc:
if config.backend == "transformers_inprocess":
status["notes"].append(f"Torch CUDA probe failed: {exc}")
try:
from transformers import AutoProcessor # noqa: F401
status["auto_processor"] = True
if config.backend == "transformers_inprocess" and status["model_path_exists"] and status["is_hf_repo"]:
try:
processor = AutoProcessor.from_pretrained(config.model_path, local_files_only=True)
status["processor_class"] = type(processor).__name__
except Exception as exc:
status["notes"].append(f"AutoProcessor load failed: {exc}")
except Exception as exc:
if config.backend == "transformers_inprocess":
status["notes"].append(f"AutoProcessor import failed: {exc}")
try:
from transformers import DiffusionGemmaForBlockDiffusion # noqa: F401
status["diffusion_gemma_class"] = True
except Exception as exc:
if config.backend == "transformers_inprocess":
status["notes"].append(f"DiffusionGemmaForBlockDiffusion import failed: {exc}")
except Exception as exc:
if config.backend == "transformers_inprocess":
status["notes"].append(f"Transformers probe failed: {exc}")
if config.backend == "transformers_inprocess":
modelopt_nvfp4 = status["path_kind"] == "nvfp4_hf_repo" and status["quant_method"] == "modelopt"
low_bridge_memory = bool(
modelopt_nvfp4
and config.quantization == "modelopt_nvfp4"
and 0 < float(config.max_memory_gb) < MIN_MODELOPT_NVFP4_BRIDGE_MEMORY_GB
)
status["nvfp4_bridge_memory_guard_passed"] = not low_bridge_memory
if modelopt_nvfp4 and config.quantization == "modelopt_nvfp4":
smoke = _nvfp4_bridge_smoke(config.model_path)
status["nvfp4_bridge_smoke"] = smoke
status["comfy_nvfp4_bridge_supported"] = bool(
smoke.get("passed")
and status["comfy_kitchen_importable"]
and status["torch_cuda_available"]
and status["diffusion_gemma_class"]
and status["auto_processor"]
)
status["modelopt_nvfp4_supported"] = bool(
modelopt_nvfp4
and (
status["comfy_nvfp4_bridge_supported"]
or (
status["modelopt_importable"]
and status["tensorrt_llm_importable"]
and status["has_modelopt_state"]
)
)
)
status["ready"] = bool(
status["model_path_exists"]
and status["auto_processor"]
and status["diffusion_gemma_class"]
and not (modelopt_nvfp4 and not status["modelopt_nvfp4_supported"])
and not low_bridge_memory
)
status["supports_pixels"] = bool(status["ready"])
status["supports_video"] = bool(status["ready"])
status["supports_thinking"] = bool(status["ready"])
status["media_policy"] = "pixels_when_ready"
if modelopt_nvfp4 and status["comfy_nvfp4_bridge_supported"]:
status["notes"].append(
"This NVFP4 repo uses quant_method=modelopt, which vanilla Transformers does not deserialize. "
"The node will use the local Comfy NVFP4 bridge: Transformers supplies DiffusionGemma structure "
"and Comfy/comfy_kitchen supplies packed NVFP4 expert matmul."
)
if low_bridge_memory:
status["notes"].append(
f"The local NVFP4 bridge does not offload by max_memory_gb yet. Set max_memory_gb to "
f"{MIN_MODELOPT_NVFP4_BRIDGE_MEMORY_GB:g} or higher, or 0 for an explicit full-GPU load. "
"Lower values are blocked to avoid a partial CUDA load that can require a ComfyUI restart."
)
elif modelopt_nvfp4 and not status["modelopt_nvfp4_supported"]:
status["notes"].append(
"This NVFP4 repo uses quant_method=modelopt. Transformers 5.12.1 exposes DiffusionGemma, "
"but it does not register modelopt as an in-process quantizer. The local ModelOpt package "
"also lacks a modelopt_state.pth restore file for this repo, TensorRT-LLM is not available, "
"and the Comfy NVFP4 bridge smoke did not pass. "
"Blocking before full weight load to avoid a CPU-memory hang."
)
elif config.backend == "gguf_subprocess":
status["cli_path_exists"] = bool(config.cli_path and Path(config.cli_path).exists())
status["cuda_runtime_bin_exists"] = Path(DEFAULT_CUDA_RUNTIME_BIN).exists()
status["ready"] = bool(status["model_path_exists"] and status["cli_path_exists"] and status["is_gguf_file"])
status["supports_pixels"] = False
status["supports_video"] = False
status["supports_thinking"] = bool(status["ready"])
status["media_policy"] = "metadata_only"
if not status["cli_path_exists"]:
status["notes"].append("GGUF subprocess backend needs a foreground llama-diffusion-cli path.")
if status["model_path_exists"] and not status["is_gguf_file"]:
if status["path_kind"] == "nvfp4_hf_repo":
status["notes"].append(
"This path is an NVFP4 Hugging Face/vLLM repo, not a .gguf file. "
"The clean Model Loader runs llama-diffusion-cli and needs the GGUF model path."
)
elif status["is_hf_repo"]:
status["notes"].append(
"This path is a Hugging Face repo folder, not a .gguf file. "
"Use Model Loader (Advanced) with transformers_inprocess only after proof gates pass."
)
else:
status["notes"].append("GGUF subprocess backend needs model_path to point directly to a .gguf file.")
elif config.backend in {"qwen_vl_placeholder", "minicpm_v_placeholder"}:
status["ready"] = False
status["notes"].append(
"Fallback backend lane is reserved for comparative benchmarking; no Qwen/MiniCPM runner is loaded by this pack yet."
)
else:
status["ready"] = config.backend == "template"
status["supports_thinking"] = False
status["media_policy"] = "template"
if config.backend == "transformers_inprocess" and not status["ready"]:
if not status["diffusion_gemma_class"]:
status["notes"].append(
"This ComfyUI venv can keep the node pack loaded, but it cannot run DiffusionGemma until "
"a compatible Transformers build exposes DiffusionGemmaForBlockDiffusion."
)
if status["path_kind"] == "nvfp4_hf_repo" and not status["comfy_nvfp4_bridge_supported"]:
status["notes"].append(
"NVFP4 was detected. Use this path only after a supported in-process ModelOpt NVFP4 loader is "
"available; the current local foreground fallback remains the GGUF llama-diffusion-cli route."
)
return status
def _strip_markdown(text: str) -> str:
text = re.sub(r"```(?:json|text|markdown)?\s*", "", text, flags=re.IGNORECASE)
text = text.replace("```", "")
text = re.sub(r"^\s{0,3}#{1,6}\s+", "", text, flags=re.MULTILINE)
return text.strip()
def _extract_delimited_final_payload(text: str) -> tuple[str, bool]:
value = str(text or "")
if not value:
return "", False
lowered = value.lower()
starts: list[tuple[int, str]] = []
for marker in FINAL_JSON_OPEN_MARKERS:
index = lowered.find(marker.lower())
if index >= 0:
starts.append((index, marker))
if not starts:
return value, False
start_index, open_marker = min(starts, key=lambda item: item[0])
payload_start = start_index + len(open_marker)
payload_end = len(value)
for marker in FINAL_JSON_CLOSE_MARKERS:
index = lowered.find(marker.lower(), payload_start)
if index >= 0:
payload_end = min(payload_end, index)
return value[payload_start:payload_end].strip(), True
def _strip_thinking(text: str) -> str:
text = re.sub(r".*?", "", text, flags=re.IGNORECASE | re.DOTALL)
text = re.sub(r"<\|thought\|>.*?<\|end_thought\|>", "", text, flags=re.IGNORECASE | re.DOTALL)
text = re.sub(r"<\|thought\|>.*?(?=<\|(?:final_json|answer|final)\|>)", "", text, flags=re.IGNORECASE | re.DOTALL)
text = re.sub(r"<\|channel\|>thought.*?(?=<\|channel\|>|$)", "", text, flags=re.IGNORECASE | re.DOTALL)
text = re.sub(r"<\|?channel\|?>\s*thought\s*", "", text, flags=re.IGNORECASE)
text = re.sub(r"<\|?channel\|?>", "", text, flags=re.IGNORECASE)
text = re.sub(r"<\|[^>]+?\|>", "", text)
return text.strip()
def _strip_cli_runtime_lines(text: str) -> str:
kept: list[str] = []
for line in text.splitlines():
if re.match(r"\s*(total time:|throughput:|diffusion step:)", line, flags=re.IGNORECASE):
continue
if re.match(r"\s*\d+\.\d+\.\d+\.\d+\s+[IWE]\s+", line):
continue
kept.append(line)
return "\n".join(kept).strip()
def _strip_generation_artifacts(text: str) -> str:
value = str(text or "")
if not value:
return ""
replacements = {
"\ufeff": "",
"\ufffd": "",
"": "",
"ç´‹": " ",
}
for artifact, replacement in replacements.items():
value = value.replace(artifact, replacement)
value = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]", " ", value)
value = re.sub(r"(?:(?<=\s)|(?<=[,.;:!?]))(?:[ÃÂâ][^\sA-Za-z0-9]{1,8}|[ÃÂ][A-Za-z0-9]{1,4})(?=\s|[,.;:!?]|$)", " ", value)
value = re.sub(r"\s*,\s+to\s+(him|her|them)\b", r" near \1", value, flags=re.IGNORECASE)
value = re.sub(r"\s+([,.!?;:])", r"\1", value)
value = re.sub(r"([,.!?;:])(?=[^\s,.!?;:])", r"\1 ", value)
value = re.sub(
r'([,.!?;:])\s+(["\'”’»])(?=\s|$|[,.;:!?)\]}])',
r"\1\2",
value,
)
return re.sub(r"\s+", " ", value).strip()
def _split_extra_args(extra_args: str) -> list[str]:
if not extra_args.strip():
return []
if os.name != "nt":
return shlex.split(extra_args)
return [part.strip("\"'") for part in shlex.split(extra_args, posix=False)]
def _format_temperature(value: Any, default: float = 0.2) -> str:
temp = max(0.0, min(2.0, _safe_float(value, default)))
return f"{temp:.3f}".rstrip("0").rstrip(".")
def _extra_args_with_temperature(extra_args: str, temperature: float) -> str:
value = (extra_args or "").strip()
temp = _format_temperature(temperature)
if not value:
return f"--temp {temp}"
pattern = r"(? dict[str, str]:
env = os.environ.copy()
path_key = "Path" if "Path" in env else "PATH"
existing_path = env.get(path_key, "")
candidates = [
str(Path(cli_path).parent) if cli_path else "",
DEFAULT_CUDA_RUNTIME_BIN,
]
prefix = [path for path in candidates if path and Path(path).exists()]
env[path_key] = os.pathsep.join([*prefix, existing_path]) if existing_path else os.pathsep.join(prefix)
cuda_root = Path(DEFAULT_CUDA_RUNTIME_BIN).parent
if cuda_root.exists():
env.setdefault("CUDA_PATH", str(cuda_root))
return env
def _sanitize_prompt_text(
text: str,
fallback: str = "",
max_chars: int = 4000,
strip_thinking: bool = True,
strip_markdown: bool = True,
) -> str:
value = (text or "").strip()
if strip_thinking:
value = _strip_thinking(value)
if strip_markdown:
value = _strip_markdown(value)
value = _strip_generation_artifacts(value)
value = re.sub(r"\s+", " ", value).strip()
if not value:
value = (fallback or "").strip()
if max_chars > 0 and len(value) > max_chars:
value = value[:max_chars].rsplit(" ", 1)[0].strip() or value[:max_chars].strip()
return value
def _sanitize_structured_prompt_text(
text: str,
fallback: str = "",
max_chars: int = 8000,
strip_thinking: bool = True,
strip_markdown: bool = True,
) -> str:
value = str(text or "").replace("\r\n", "\n").replace("\r", "\n").strip()
if strip_thinking:
value = _strip_thinking(value)
if strip_markdown:
value = _strip_markdown(value)
replacements = {
"\ufeff": "",
"\ufffd": "",
"": "",
}
for artifact, replacement in replacements.items():
value = value.replace(artifact, replacement)
value = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]", " ", value)
lines = []
for line in value.splitlines():
line = re.sub(r"[ \t]+", " ", line).strip()
lines.append(line)
value = "\n".join(lines)
value = re.sub(r"\n{3,}", "\n\n", value).strip()
if not value:
value = str(fallback or "").strip()
if max_chars > 0 and len(value) > max_chars:
value = value[:max_chars].rstrip()
return value
_MINIMAX_H3_REF_FIELDS = (
"subject_definitions",
"summary",
"retention_analysis",
"detailed_description",
"overall_soundscape",
"non_diegetic_music",
)
def _minimax_h3_ref_sections(prompt: str) -> dict[str, str] | None:
pattern = r"(?ms)^" + r"\s*\n+\s*".join(
rf"{re.escape(field_name)}\s*:\s*(.*?)" for field_name in _MINIMAX_H3_REF_FIELDS[:-1]
) + rf"\s*\n+\s*{re.escape(_MINIMAX_H3_REF_FIELDS[-1])}\s*:\s*(.*)$"
match = re.fullmatch(pattern, prompt)
if not match:
return None
return {
field_name: match.group(index + 1).strip()
for index, field_name in enumerate(_MINIMAX_H3_REF_FIELDS)
}
def _rebuild_minimax_h3_ref_prompt(sections: dict[str, str]) -> str:
return "\n\n".join(f"{field_name}:\n{sections.get(field_name, '').strip()}" for field_name in _MINIMAX_H3_REF_FIELDS)
def _normalize_minimax_h3_prompt(text: str, minimax_h3_mode: str = "t2va") -> str:
prompt = _sanitize_structured_prompt_text(text, "", 0)
if not prompt:
return ""
# These are unambiguous transport defects, not creative edits. Host-only
# grounding annotations must never reach H3, and DiffusionGemma sometimes
# drops the leading N from the exact N/A absence sentinel.
prompt = _strip_grounding_role_annotations(prompt)
prompt = re.sub(
r"(?im)^(?P[ \t]*(?:overall_soundscape|non_diegetic_music)[ \t]*:[ \t]*(?:\n[ \t]*)?)/A(?=[ \t]*(?:$|\n))",
lambda match: f"{match.group('prefix')}N/A",
prompt,
)
if _normalize_minimax_h3_mode(minimax_h3_mode) == "ref2va":
# DiffusionGemma occasionally drops the leading "n" from this one
# required heading. The line-anchored alias is unambiguous and only
# changes the field label; its generated music guidance is preserved.
music_alias_pattern = re.compile(r"(?im)^\s*on[ _-]*diegetic[ _-]*music\s*:")
canonical_music_pattern = re.compile(
r"(?im)^\s*non[ _-]*diegetic[ _-]*music\s*:"
)
music_alias_matches = list(music_alias_pattern.finditer(prompt))
canonical_music_matches = list(canonical_music_pattern.finditer(prompt))
preceding_heading_positions: list[int] = []
preceding_headings_are_canonical = True
for field_name in _MINIMAX_H3_REF_FIELDS[:-1]:
field_matches = list(re.finditer(rf"(?im)^\s*{re.escape(field_name)}\s*:", prompt))
if len(field_matches) != 1:
preceding_headings_are_canonical = False
break
preceding_heading_positions.append(field_matches[0].start())
preceding_structure_is_canonical = bool(
preceding_headings_are_canonical
and preceding_heading_positions == sorted(preceding_heading_positions)
)
# A common constrained-decoding miss is a duplicated, empty final music
# heading. The first section is complete, so dropping only the empty EOF
# duplicate is lossless and avoids sending the same packet through two
# expensive model repair attempts.
if (
preceding_structure_is_canonical
and len(canonical_music_matches) == 2
and canonical_music_matches[0].start() > preceding_heading_positions[-1]
and canonical_music_matches[1].start() > canonical_music_matches[0].start()
and prompt[
canonical_music_matches[0].end() : canonical_music_matches[1].start()
].strip()
and not prompt[canonical_music_matches[1].end() :].strip()
):
prompt = prompt[: canonical_music_matches[1].start()].rstrip()
canonical_music_matches = list(canonical_music_pattern.finditer(prompt))
alias_has_nonempty_body = bool(
len(music_alias_matches) == 1
and (
prompt[music_alias_matches[0].end() :].strip()
if not canonical_music_matches
else music_alias_matches[0].start() < canonical_music_matches[0].start()
and prompt[
music_alias_matches[0].end() : canonical_music_matches[0].start()
].strip()
)
)
empty_canonical_footer_after_alias = bool(
len(music_alias_matches) == 1
and len(canonical_music_matches) == 1
and music_alias_matches[0].start() < canonical_music_matches[0].start()
and not prompt[canonical_music_matches[0].end() :].strip()
)
if (
preceding_structure_is_canonical
and len(music_alias_matches) == 1
and music_alias_matches[0].start() > preceding_heading_positions[-1]
and alias_has_nonempty_body
and (not canonical_music_matches or empty_canonical_footer_after_alias)
):
if empty_canonical_footer_after_alias:
prompt = prompt[: canonical_music_matches[0].start()].rstrip()
prompt = music_alias_pattern.sub("non_diegetic_music:", prompt, count=1)
prompt = re.sub(
r"<\s*(subject|picture|image|video|audio)\s+(\d+)\s*>",
lambda match: (
f""
if match.group(1).casefold() == "subject"
else f"<{'Picture' if match.group(1).casefold() in {'picture', 'image'} else match.group(1).title()} {int(match.group(2))}>"
),
prompt,
flags=re.IGNORECASE,
)
for field_name in _MINIMAX_H3_REF_FIELDS:
prompt = re.sub(
rf"(?im)^\s*{re.escape(field_name)}\s*:",
f"{field_name}:",
prompt,
)
sections = _minimax_h3_ref_sections(prompt)
return _rebuild_minimax_h3_ref_prompt(sections) if sections else prompt
fields = (
"integrated_multimodal_description",
"overall_soundscape",
"non_diegetic_music",
)
for field_name in fields:
prompt = re.sub(
rf"(?im)^\s*{re.escape(field_name)}\s*:",
f"{field_name}:",
prompt,
)
field_matches = {
field_name: re.search(rf"(?m)^{re.escape(field_name)}\s*:", prompt)
for field_name in fields
}
if not any(field_matches.values()) and re.match(r"^\[Shot 1\](?=\s|$)", prompt):
blocks = [block.strip() for block in re.split(r"\n\s*\n", prompt) if block.strip()]
if len(blocks) >= 3:
integrated = "\n\n".join(blocks[:-2])
prompt = (
f"integrated_multimodal_description: {integrated}\n\n"
f"overall_soundscape: {blocks[-2]}\n\n"
f"non_diegetic_music: {blocks[-1]}"
)
elif (
not field_matches["integrated_multimodal_description"]
and field_matches["overall_soundscape"]
and field_matches["non_diegetic_music"]
and re.match(r"^\[Shot 1\](?=\s|$)", prompt)
):
prompt = f"integrated_multimodal_description: {prompt}"
structured_match = re.fullmatch(
r"(?ms)integrated_multimodal_description\s*:\s*(.*?)\s*\n+\s*overall_soundscape\s*:\s*(.*?)\s*\n+\s*non_diegetic_music\s*:\s*(.*)",
prompt,
)
if structured_match:
integrated = structured_match.group(1).strip()
shot_labels = re.findall(r"\[Shot (\d+)\]", integrated, flags=re.IGNORECASE)
if shot_labels == ["1"]:
next_shot = 1
def label_unmarked_cut(match: re.Match[str]) -> str:
nonlocal next_shot
next_shot += 1
return f"[Shot {next_shot}] {match.group(0)}"
integrated = re.sub(
r"\bAt\s+\d{2,}:\d{2}\.\d{3},?\s+(?=the\s+(?:camera|shot|scene)\s+(?:(?:hard\s+)?cuts|transitions|changes|switches|cross-dissolves|fades|wipes)\s+(?:to|into)\b)",
label_unmarked_cut,
integrated,
flags=re.IGNORECASE,
)
prompt = (
f"integrated_multimodal_description: {integrated}\n\n"
f"overall_soundscape: {structured_match.group(2).strip()}\n\n"
f"non_diegetic_music: {structured_match.group(3).strip()}"
)
return re.sub(
r"(\[Shot (?:[2-9]|[1-9]\d+)\]\s+At\s+\d{2,}:\d{2}\.\d{3})(?!,)(?=\s+the\s+(?:camera|shot|scene)\b)",
r"\1,",
prompt,
flags=re.IGNORECASE,
)
def _minimax_h3_ref_definition_items(text: str) -> list[tuple[str, str]]:
return [
(f"<{kind.title()} {int(ordinal)}>", description.strip())
for kind, ordinal, description in re.findall(
r"(?im)^[ \t]*<[ \t]*(Subject|Picture|Video|Audio)[ \t]+(\d+)[ \t]*>"
r"[ \t]*(?::|=|[-–—]|\bis\b)[ \t]*(\S.*)$",
text or "",
)
]
def _minimax_h3_ref2va_subject_count(prompt_text: Any) -> int:
sections = _minimax_h3_ref_sections(
_sanitize_structured_prompt_text(str(prompt_text or ""), "", 0)
)
if sections is None:
return 0
return len(
{
tag
for tag, _description in _minimax_h3_ref_definition_items(
sections["subject_definitions"]
)
if tag.startswith(" list[str]:
manifest = _normalize_minimax_h3_reference_manifest(reference_manifest)
definitions = _minimax_h3_reference_definitions(manifest)
tags = [tag for tag, _description in definitions]
combined = f"{user_prompt} {manifest}"
task_types: list[str] = []
if _has_unnegated_prompt_phrase(
combined,
r"\b(?:keyframes?|first[-\s]+frames?|last[-\s]+frames?|first\s+and\s+last\s+frames?|interpolat(?:e|ion|ing))\b",
):
task_types.append("keyframe completion")
if any(tag.startswith((" str:
"""Fill only unusable later-shot cut instants on a bounded local clock.
Valid authored cut instants are retained. When authored instants conflict,
the maximum-size increasing subset is kept and only the unusable gaps are
interpolated. Shot labels, shot bodies, and shot count are never changed.
"""
duration_milliseconds = int(
round(max(0.0, _safe_float(duration_seconds, 0.0)) * 1000.0)
)
shot_pattern = re.compile(r"\[Shot (?P\d+)\]")
shots = list(shot_pattern.finditer(detailed or ""))
shot_numbers = [int(match.group("shot")) for match in shots]
shot_count = len(shots)
if (
duration_milliseconds <= shot_count
or shot_count <= 1
or shot_numbers != list(range(1, shot_count + 1))
):
return detailed
authored_prefix_pattern = re.compile(
r"\s+At\s+(?P\d{2,}):(?P\d{2})\."
r"(?P\d{3})(?P,?)",
flags=re.IGNORECASE,
)
validator_prefix_pattern = re.compile(
r"\s+At\s+(?P\d{2,}):(?P\d{2})\."
r"(?P\d{3}),"
)
candidates: list[tuple[int, int]] = []
prefix_matches: dict[int, re.Match[str]] = {}
canonical_prefixes: dict[int, bool] = {}
for shot_index, marker in enumerate(shots[1:], start=2):
authored = authored_prefix_pattern.match(detailed, marker.end())
if authored is None:
continue
prefix_matches[shot_index] = authored
canonical_prefixes[shot_index] = bool(
validator_prefix_pattern.match(detailed, marker.end())
)
seconds = int(authored.group("seconds"))
if seconds >= 60:
continue
authored_milliseconds = (
int(authored.group("minutes")) * 60_000
+ seconds * 1000
+ int(authored.group("milliseconds"))
)
# An explicit authored cut at or beyond the clip boundary is a
# semantic scheduling error, not a missing/local-format defect. Keep
# it intact so the strict validator reports the distinct
# ``minimax_h3_cut_timestamp_out_of_range`` reason and the established
# refinement path can re-plan it.
if authored_milliseconds >= duration_milliseconds:
return detailed
# Leave one millisecond for every missing cut and for the terminal
# video boundary. H3 durations are much wider than this guard, but it
# keeps interpolation exact after integer rounding.
if (
authored_milliseconds < shot_index - 1
or duration_milliseconds - authored_milliseconds
< shot_count - shot_index + 1
):
continue
candidates.append((shot_index, authored_milliseconds))
# Longest increasing feasible subsequence: retain as many authored cut
# instants as possible. Equal-length ties prefer earlier shot anchors.
paths: list[list[tuple[int, int]]] = []
for shot_index, authored_milliseconds in candidates:
best_path: list[tuple[int, int]] = []
for previous_path in paths:
previous_shot, previous_milliseconds = previous_path[-1]
if (
previous_shot < shot_index
and authored_milliseconds - previous_milliseconds
>= shot_index - previous_shot
and len(previous_path) > len(best_path)
):
best_path = previous_path
paths.append([*best_path, (shot_index, authored_milliseconds)])
selected_path: list[tuple[int, int]] = []
if paths:
maximum_length = max(len(path) for path in paths)
selected_path = min(
(path for path in paths if len(path) == maximum_length),
key=lambda path: tuple(shot for shot, _milliseconds in path),
)
anchors = [(1, 0), *selected_path, (shot_count + 1, duration_milliseconds)]
repaired_times: dict[int, int] = {
shot_index: authored_milliseconds
for shot_index, authored_milliseconds in selected_path
}
for (start_shot, start_milliseconds), (end_shot, end_milliseconds) in zip(
anchors,
anchors[1:],
):
shot_span = end_shot - start_shot
time_span = end_milliseconds - start_milliseconds
for shot_index in range(start_shot + 1, end_shot):
offset = shot_index - start_shot
repaired_times[shot_index] = start_milliseconds + (
time_span * offset + shot_span // 2
) // shot_span
replacements: list[tuple[int, int, str]] = []
selected_shots = {shot_index for shot_index, _milliseconds in selected_path}
for shot_index, marker in enumerate(shots[1:], start=2):
cut_milliseconds = repaired_times[shot_index]
minutes, remainder = divmod(cut_milliseconds, 60_000)
seconds, milliseconds = divmod(remainder, 1000)
replacement = f" At {minutes:02d}:{seconds:02d}.{milliseconds:03d},"
authored = prefix_matches.get(shot_index)
if (
shot_index in selected_shots
and canonical_prefixes.get(shot_index, False)
):
continue
if authored is None:
replacements.append((marker.end(), marker.end(), replacement))
else:
replacements.append((authored.start(), authored.end(), replacement))
repaired = detailed
for start, end, replacement in sorted(replacements, reverse=True):
repaired = repaired[:start] + replacement + repaired[end:]
return repaired
def _repair_minimax_h3_timed_cut_openers(timeline: str) -> str:
"""Canonicalize only timestamp-adjacent H3 cut transport syntax.
The model-authored shot body remains byte-for-byte intact. When a later
shot starts directly with action or camera motion, a neutral cut prefix is
inserted so native H3 receives the delimiter its parser requires.
"""
repaired = str(timeline or "")
# The live T2VA failure omitted the comma only before an otherwise
# unmistakable smash-cut opener (``At 00:06.000 the smash cuts to``).
# Restore punctuation only for that exact transition family; ordinary
# timestamp-adjacent prose remains untouched and will still fail closed.
repaired = re.sub(
r"(\[Shot (?:[2-9]|[1-9]\d+)\]\s+At\s+\d{2,}:\d{2}\.\d{3})(?!,)"
r"(?=\s+(?:the\s+)?(?:a\s+)?smash(?:[- ]+cut|[- ]?cuts?)\s+to\b)",
r"\1,",
repaired,
flags=re.IGNORECASE,
)
timestamp_prefix = (
r"(\[Shot (?:[2-9]|[1-9]\d+)\]\s+At\s+\d{2,}:\d{2}\.\d{3},\s*)"
)
repaired = re.sub(
timestamp_prefix
+ r"the\s+camera\s+smash(?:[- ]+cut|[- ]?cuts?)\s+to\s*:?\s*",
r"\1the camera hard cuts to ",
repaired,
flags=re.IGNORECASE,
)
repaired = re.sub(
timestamp_prefix
+ r"(?:the\s+)?(?:a\s+)?smash(?:[- ]+cut|[- ]?cuts?)\s+to\s*:?\s*",
r"\1the camera hard cuts to ",
repaired,
flags=re.IGNORECASE,
)
repaired = re.sub(
timestamp_prefix
+ r"the\s+camera\s+(?:quick|clean)\s+cuts?\s+to\s+",
r"\1the camera cuts to ",
repaired,
flags=re.IGNORECASE,
)
repaired = re.sub(
timestamp_prefix + r"(?:a\s+)?hard\s+cut\s+to\s*:?\s*",
r"\1the camera hard cuts to ",
repaired,
flags=re.IGNORECASE,
)
repaired = re.sub(
timestamp_prefix + r"(?:a\s+)?(?:quick|clean)?\s*cut\s+to\s*:?\s*",
r"\1the camera cuts to ",
repaired,
flags=re.IGNORECASE,
)
return re.sub(
r"(\[Shot (?:[2-9]|[1-9]\d+)\]\s+At\s+\d{2,}:\d{2}\.\d{3},)"
r"(?!\s*the\s+(?:camera|shot|scene)\s+(?:(?:hard\s+)?cuts|transitions|changes|switches|cross-dissolves|fades|wipes)\s+(?:to|into)\b)"
r"(\s*)",
r"\1\2the shot cuts to: ",
repaired,
flags=re.IGNORECASE,
)
def _repair_minimax_h3_ref_detailed_timeline(detailed: str, duration_seconds: float = 0.0) -> str:
range_pattern = re.compile(
r"\[Shot (?P\d+)\]\s*\(?\s*"
r"(?P\d{1,2}):(?P\d{2})(?:\.(?P\d{1,3}))?\s*[-–—]\s*"
r"(?P\d{1,2}):(?P\d{2})(?:\.(?P\d{1,3}))?\s*\)?\s*:?\s*",
flags=re.IGNORECASE,
)
range_matches = list(range_pattern.finditer(detailed))
duration = max(0.0, _safe_float(duration_seconds, 0.0))
def range_time(match: re.Match[str], prefix: str, mode: str) -> float:
major = int(match.group(f"{prefix}_major"))
minor = int(match.group(f"{prefix}_minor"))
fraction_text = match.group(f"{prefix}_fraction") or ""
if mode == "minutes_seconds":
if minor >= 60:
return -1.0
fraction = int(fraction_text.ljust(3, "0")) / 1000.0 if fraction_text else 0.0
return major * 60.0 + minor + fraction
if fraction_text or minor >= 100:
return -1.0
return major + minor / 100.0
def valid_range_mode(mode: str) -> bool:
previous_end: float | None = None
for match in range_matches:
start = range_time(match, "start", mode)
end = range_time(match, "end", mode)
if start < 0 or end <= start:
return False
if int(match.group("shot")) == 1 and start != 0:
return False
if previous_end is not None and abs(start - previous_end) > 0.01:
return False
if duration > 0 and (start >= duration or end > duration + 0.05):
return False
previous_end = end
return bool(range_matches)
range_mode = ""
if valid_range_mode("minutes_seconds"):
range_mode = "minutes_seconds"
elif duration > 0 and valid_range_mode("seconds_hundredths"):
range_mode = "seconds_hundredths"
unsafe_explicit_range = bool(range_matches) and not range_mode
if range_mode:
def replace_range(match: re.Match[str]) -> str:
shot = int(match.group("shot"))
start = range_time(match, "start", range_mode)
if shot == 1:
return "[Shot 1] "
minutes = int(start // 60)
seconds = int(start % 60)
milliseconds = int(round((start - int(start)) * 1000))
if milliseconds == 1000:
seconds += 1
milliseconds = 0
if seconds == 60:
minutes += 1
seconds = 0
following = match.string[match.end() :]
article = "a " if re.match(
r"(?:close[- ]up|medium(?:[- ]wide)? shot|wide shot|long shot|high[- ]angle|low[- ]angle|bird['’]s[- ]eye)",
following,
flags=re.IGNORECASE,
) else ""
return f"[Shot {shot}] At {minutes:02d}:{seconds:02d}.{milliseconds:03d}, the camera cuts to {article}"
detailed = range_pattern.sub(replace_range, detailed)
detailed = re.sub(
r"(\[Shot (?:[2-9]|[1-9]\d+)\]\s+At\s+\d{2,}:\d{2}\.\d{3},\s*the camera cuts to (?:a )?)([A-Z])",
lambda match: f"{match.group(1)}{match.group(2).casefold()}",
detailed,
)
seconds_range_pattern = re.compile(
r"\[Shot (?P\d+)\]\s*\(?\s*"
r"(?P\d+(?:\.\d{1,3})?)\s*s?\s*[-–—]\s*"
r"(?P\d+(?:\.\d{1,3})?)\s*s\s*\)?\s*:?\s*",
flags=re.IGNORECASE,
)
seconds_range_matches = list(seconds_range_pattern.finditer(detailed))
def valid_seconds_ranges() -> bool:
previous_end: float | None = None
for match in seconds_range_matches:
start = float(match.group("start"))
end = float(match.group("end"))
if end <= start:
return False
if int(match.group("shot")) == 1 and abs(start) > 0.001:
return False
if previous_end is not None and abs(start - previous_end) > 0.01:
return False
if duration > 0 and (start >= duration or end > duration + 0.05):
return False
previous_end = end
return bool(seconds_range_matches)
seconds_ranges_valid = valid_seconds_ranges()
unsafe_explicit_range = unsafe_explicit_range or (
bool(seconds_range_matches) and not seconds_ranges_valid
)
if seconds_ranges_valid:
def replace_seconds_range(match: re.Match[str]) -> str:
shot = int(match.group("shot"))
start = float(match.group("start"))
if shot == 1:
return "[Shot 1] "
total_milliseconds = int(round(start * 1000.0))
minutes, remainder = divmod(total_milliseconds, 60_000)
seconds, milliseconds = divmod(remainder, 1000)
following = match.string[match.end() :]
article = "a " if re.match(
r"(?:close[- ]up|medium(?:[- ]wide)? shot|wide shot|long shot|high[- ]angle|low[- ]angle|bird['’]s[- ]eye)",
following,
flags=re.IGNORECASE,
) else ""
return f"[Shot {shot}] At {minutes:02d}:{seconds:02d}.{milliseconds:03d}, the camera cuts to {article}"
detailed = seconds_range_pattern.sub(replace_seconds_range, detailed)
detailed = re.sub(
r"(\[Shot (?:[2-9]|[1-9]\d+)\]\s+At\s+\d{2,}:\d{2}\.\d{3},\s*the camera cuts to (?:a )?)([A-Z])",
lambda match: f"{match.group(1)}{match.group(2).casefold()}",
detailed,
)
detailed = re.sub(
r"(\[Shot 1\])\s+At\s+00:00\.000,?\s*",
r"\1 ",
detailed,
flags=re.IGNORECASE,
)
if not unsafe_explicit_range:
detailed = _repair_minimax_h3_ref_cut_timestamps(detailed, duration)
detailed = re.sub(
r"(\[Shot (?:[2-9]|[1-9]\d+)\]\s+At\s+\d{2,}:\d{2}\.\d{3})(?!,)(?=\s+\S)",
r"\1,",
detailed,
flags=re.IGNORECASE,
)
return _repair_minimax_h3_timed_cut_openers(detailed)
def _repair_minimax_h3_ref_subject_timeline_labels(sections: dict[str, str]) -> None:
detailed = sections["detailed_description"]
shot1_match = re.search(r"\[Shot 1\]", detailed, flags=re.IGNORECASE)
if not shot1_match:
return
retention = sections["retention_analysis"]
timeline = detailed[shot1_match.start() :]
missing_subjects = [
(tag, description)
for tag, description in _minimax_h3_ref_definition_items(
sections["subject_definitions"]
)
if tag.startswith("\S.*)$",
retention,
)
if retention_match is None:
return
subject_semantics = f"{description} {retention_match.group('body')}"
if not re.search(
r"\b(?:environment|landscape|setting|world|location|background|scenery|production design|visual style|palette|"
r"lighting|town|city|street|square|field|forest|room|interior|exterior)\b|"
r"^\s*(?:the\s+|an?\s+)?(?:void\s+of\s+space|(?:outer|deep)\s+space|"
r"(?:curved\s+)?horizon\s+of\s+(?:the\s+)?earth|earth(?:['’]s)?\s+(?:curved\s+)?(?:horizon|curve)|"
r"(?:orbital|planetary)\s+(?:space|vista|scene)|starfield|starscape|night\s+sky)\b",
subject_semantics,
flags=re.IGNORECASE,
) or re.search(
r"\b(?:person|character|courier|protagonist|antagonist|creature|performer|actor|man|woman|boy|girl|animal|monster)\b",
description,
flags=re.IGNORECASE,
):
return
source_tags = _minimax_h3_reference_tags(description)
if len(source_tags) == 1 and source_tags[0] in timeline:
repairs.append((tag, source_tags[0]))
continue
if (
len(missing_subjects) == 1
and not source_tags
and re.search(
r"\b(?:environment|landscape|setting|world|location|background|scenery|town|city|street|square|field|"
r"forest|room|interior|exterior)\b|"
r"^\s*(?:the\s+|an?\s+)?(?:void\s+of\s+space|(?:outer|deep)\s+space|"
r"(?:curved\s+)?horizon\s+of\s+(?:the\s+)?earth|earth(?:['’]s)?\s+(?:curved\s+)?(?:horizon|curve)|"
r"(?:orbital|planetary)\s+(?:space|vista|scene)|starfield|starscape|night\s+sky)\b",
subject_semantics,
flags=re.IGNORECASE,
)
):
implicit_environment_tag = tag
continue
return
if repairs and len({source_tag for _tag, source_tag in repairs}) != len(repairs):
return
for tag, source_tag in repairs:
timeline = timeline.replace(source_tag, f"{tag} (derived from {source_tag})", 1)
if implicit_environment_tag:
relative_end = shot1_match.end() - shot1_match.start()
suffix = timeline[relative_end:]
separator = "" if suffix.startswith((" ", "\n")) else " "
timeline = (
f"{timeline[:relative_end]} The setting is {implicit_environment_tag} as defined above."
f"{separator}{suffix}"
)
if not repairs and not implicit_environment_tag:
return
sections["detailed_description"] = f"{detailed[: shot1_match.start()]}{timeline}"
def _repair_minimax_h3_ref_retention_rows(
sections: dict[str, str],
definition_items: list[tuple[str, str]] | None = None,
reference_manifest: str = "",
) -> None:
"""Canonicalize only an incomplete or invalid Ref2VA retention table."""
items = list(
definition_items
if definition_items is not None
else _minimax_h3_ref_definition_items(sections.get("subject_definitions", ""))
)
if not items:
return
retention = sections.get("retention_analysis", "")
manifest_definitions = dict(
_minimax_h3_reference_definitions(
_normalize_minimax_h3_reference_manifest(reference_manifest)
)
)
def allowed_markers(kind: str) -> set[str]:
return (
_MINIMAX_H3_REF_AUDIO_RETENTION
if kind == "Audio"
else _MINIMAX_H3_REF_VISUAL_RETENTION
)
def default_marker(
kind: str,
tag: str,
definition: str,
authored_marker: str = "",
) -> str:
if kind == "Subject":
return "fully_preserved"
if kind != "Audio":
return "attribute_transfer"
if authored_marker == "fully_preserved":
return "fully_copy"
if authored_marker == "partially_preserved":
return "partially_copy"
declared_semantics = (
f"{definition} {manifest_definitions.get(tag, '')}"
)
if re.search(
r"\b(?:exact|locked|copy|copied|reuse|reused|retain|retained|preserve|preserved)\b",
declared_semantics,
flags=re.IGNORECASE,
):
return "fully_copy"
return "reference"
all_rows_valid = True
for tag, _definition in items:
kind_match = re.fullmatch(r"<(Subject|Picture|Video|Audio) \d+>", tag)
if kind_match is None:
all_rows_valid = False
break
marker_pattern = "|".join(
re.escape(marker) for marker in sorted(allowed_markers(kind_match.group(1)))
)
if not re.search(
rf"(?m)^\s*{re.escape(tag)}(?:\s*\([^\n)]*\))?\s*:\s*"
rf"(?:{marker_pattern})\s+-\s*\S",
retention,
):
all_rows_valid = False
break
if all_rows_valid:
return
retention_lines: list[str] = []
all_marker_pattern = (
r"fully_preserved|partially_preserved|attribute_transfer|weak_reference|"
r"fully_copy|partially_copy|reference"
)
for tag, definition in items:
tag_match = re.fullmatch(r"<(Subject|Picture|Video|Audio) (\d+)>", tag)
if tag_match is None:
continue
kind, ordinal = tag_match.groups()
existing_match = re.search(
rf"(?is)(?:<|\[)\s*{kind}\s+{ordinal}\s*(?:>|\])"
rf"(?:\s*(?:\([^\n)]*\)|\[[^\n\]]*\]))*\s*:\s*"
r"(.*?)(?=(?:<|\[)\s*(?:Subject|Picture|Video|Audio)\s+\d+\s*(?:>|\])|$)",
retention,
)
existing = (
existing_match.group(1).strip().strip("[]").strip()
if existing_match
else ""
)
marker_match = re.match(
rf"^({all_marker_pattern})\s*-\s*(.*)$",
existing,
flags=re.IGNORECASE | re.DOTALL,
)
if marker_match:
marker = marker_match.group(1).casefold()
retention_detail = marker_match.group(2).strip().strip("[]").strip()
else:
marker = default_marker(kind, tag, definition)
retention_detail = re.sub(
r"^[A-Za-z][A-Za-z_]{1,31}\s*-\s*",
"",
re.sub(r"^[^:]{0,48}:\s*", "", existing),
).strip().strip("[]").strip()
if marker not in allowed_markers(kind):
marker = default_marker(kind, tag, definition, marker)
retention_lines.append(
f"{tag}: {marker} - {retention_detail or definition}"
)
if retention_lines:
sections["retention_analysis"] = "\n".join(retention_lines)
def _repair_minimax_h3_ref_manifest_definitions(
sections: dict[str, str],
reference_manifest: str = "",
) -> bool:
"""Restore only host-declared reference assets omitted by the model.
The reference manifest is a validated workflow contract, so copying one of
its missing Picture/Video/Audio declarations is deterministic contract
completion rather than inferred creative content. Existing authored
definitions remain byte-stable, and unknown prompt tags still fail strict
validation.
"""
subject_definitions = str(sections.get("subject_definitions", "") or "").strip()
declared_tags = set(_minimax_h3_reference_tags(subject_definitions))
missing_lines: list[str] = []
for tag, description in _minimax_h3_reference_definitions(reference_manifest):
if tag in declared_tags:
continue
clean_description = _strip_grounding_role_annotations(description).strip(" []")
if not clean_description:
continue
missing_lines.append(f"{tag}: {clean_description}")
declared_tags.add(tag)
if not missing_lines:
return False
sections["subject_definitions"] = "\n".join(
item for item in (subject_definitions, *missing_lines) if item
)
return True
def _repair_minimax_h3_single_subject_timeline_label(
sections: dict[str, str],
) -> None:
"""Restore one unambiguous semantic Subject label without guessing roles."""
subject_items = [
(tag, description)
for tag, description in _minimax_h3_ref_definition_items(
sections.get("subject_definitions", "")
)
if tag.startswith(" bool:
"""Recover a generated timeline placed after retention rows.
DiffusionGemma can omit only the ``detailed_description:`` heading while
still authoring the style opening and every [Shot N] block. The generic
section parser then assigns that tail to ``retention_analysis``; rebuilding
the retention table used to discard the complete timeline. Move only an
unambiguous post-table tail and otherwise leave the prompt fail-closed.
"""
if str(sections.get("detailed_description", "") or "").strip():
return False
retention = str(sections.get("retention_analysis", "") or "")
shot1_match = re.search(r"\[Shot 1\]", retention, flags=re.IGNORECASE)
if shot1_match is None:
return False
marker_pattern = "|".join(
re.escape(marker)
for marker in sorted(
_MINIMAX_H3_REF_VISUAL_RETENTION | _MINIMAX_H3_REF_AUDIO_RETENTION,
key=len,
reverse=True,
)
)
row_pattern = re.compile(
rf"(?im)^[ \t]*<\s*(?:Subject|Picture|Video|Audio)\s+[1-9]\d*\s*>"
rf"(?:[ \t]*(?:\([^\n)]*\)|\[[^\n\]]*\]))?[ \t]*:[ \t]*"
rf"(?:{marker_pattern})[ \t]+-[ \t]+\S[^\r\n]*$"
)
rows = [match for match in row_pattern.finditer(retention) if match.end() <= shot1_match.start()]
if not rows:
return False
last_row_end = max(match.end() for match in rows)
detailed_tail = retention[last_row_end:].strip()
if not detailed_tail or re.search(r"\[Shot 1\]", detailed_tail, flags=re.IGNORECASE) is None:
return False
sections["retention_analysis"] = retention[:last_row_end].strip()
sections["detailed_description"] = detailed_tail
return True
def _repair_minimax_h3_ref_single_picture_subject_definition(
sections: dict[str, str],
) -> bool:
"""Promote one authored Picture definition to one used semantic Subject.
This is intentionally narrow: there must be no parsed Subject definition,
exactly one Subject tag used by summary/timeline, and exactly one Picture
definition with a substantive, identity-bearing authored description.
Environment/style plates are never promoted into performers. Multi-subject
or otherwise ambiguous prompts remain blocked for model repair.
"""
definition_items = _minimax_h3_ref_definition_items(
sections.get("subject_definitions", "")
)
if any(tag.startswith(""
for ordinal in re.findall(
r"<\s*Subject\s+([1-9]\d*)\s*>",
used_text,
flags=re.IGNORECASE,
)
}
picture_items = [
(tag, description)
for tag, description in definition_items
if tag.startswith(" tuple[str, list[str]]:
"""Insert only omitted colons in otherwise exact Ref2VA contract lines."""
prompt = str(prompt_text or "")
repairs: list[str] = []
def repair_definition_body(body: str) -> str:
definition_pattern = re.compile(
r"(?m)^(?P[ \t]*<\s*(?:Subject|Picture|Video|Audio)\s+[1-9]\d*\s*>)"
r"(?P[ \t]+)(?P\S.*)$",
flags=re.IGNORECASE,
)
def insert_definition_colon(match: re.Match[str]) -> str:
description = match.group("description")
if re.match(r"(?:[:=]|[-–—](?:\s|$)|is\b)", description, flags=re.IGNORECASE):
return match.group(0)
if len(re.findall(r"[A-Za-z0-9]{2,}", description)) < 4:
return match.group(0)
canonical_tag = re.sub(r"\s+", " ", match.group("tag").strip())
repairs.append(f"inserted_ref_definition_colon:{canonical_tag}")
return f"{match.group('tag')}:{match.group('gap')}{description}"
return definition_pattern.sub(insert_definition_colon, body)
all_retention_markers = sorted(
_MINIMAX_H3_REF_VISUAL_RETENTION | _MINIMAX_H3_REF_AUDIO_RETENTION,
key=len,
reverse=True,
)
marker_pattern = "|".join(re.escape(marker) for marker in all_retention_markers)
def repair_retention_body(body: str) -> str:
retention_pattern = re.compile(
rf"(?m)^(?P[ \t]*<\s*(?PSubject|Picture|Video|Audio)\s+[1-9]\d*\s*>"
rf"(?:[ \t]*(?:\([^\n)]*\)|\[[^\n\]]*\]))?)(?P[ \t]+)(?P{marker_pattern})"
r"(?P[ \t]+-[ \t]+\S.*)$",
flags=re.IGNORECASE,
)
def insert_retention_colon(match: re.Match[str]) -> str:
marker = match.group("marker").casefold()
allowed = (
_MINIMAX_H3_REF_AUDIO_RETENTION
if match.group("kind").casefold() == "audio"
else _MINIMAX_H3_REF_VISUAL_RETENTION
)
if marker not in allowed:
return match.group(0)
canonical_tag_match = re.search(
r"<\s*(Subject|Picture|Video|Audio)\s+([1-9]\d*)\s*>",
match.group("tag"),
flags=re.IGNORECASE,
)
canonical_tag = (
f"<{canonical_tag_match.group(1).title()} {int(canonical_tag_match.group(2))}>"
if canonical_tag_match
else match.group("tag").strip()
)
repairs.append(f"inserted_ref_retention_colon:{canonical_tag}")
return (
f"{match.group('tag')}:{match.group('gap')}{match.group('marker')}"
f"{match.group('suffix')}"
)
return retention_pattern.sub(insert_retention_colon, body)
section_specs = (
("subject_definitions", "summary", repair_definition_body),
("retention_analysis", "detailed_description", repair_retention_body),
)
replacements: list[tuple[int, int, str]] = []
for field_name, next_field, repair_body in section_specs:
match = re.search(
rf"(?ms)(?:\A|\n\n){re.escape(field_name)}\s*:\s*\n"
rf"(?P.*?)\n\n{re.escape(next_field)}\s*:",
prompt,
)
if match is None:
continue
body = match.group("body")
repaired_body = repair_body(body)
if repaired_body != body:
replacements.append((match.start("body"), match.end("body"), repaired_body))
for start, end, replacement in sorted(replacements, reverse=True):
prompt = prompt[:start] + replacement + prompt[end:]
return prompt, list(dict.fromkeys(repairs))
def _repair_minimax_h3_ref2va_structure(
prompt_text: str,
user_prompt: str = "",
reference_manifest: str = "",
duration_seconds: float = 0.0,
) -> str:
prompt = _normalize_minimax_h3_prompt(prompt_text, "ref2va")
prompt, _contract_delimiter_repairs = _repair_minimax_h3_ref2va_contract_delimiters(
prompt
)
if not prompt:
return ""
timeline_duration = max(
0.0,
_safe_float(duration_seconds, 0.0) or _explicit_prompt_duration_seconds(user_prompt),
)
canonical_sections = _minimax_h3_ref_sections(prompt)
if canonical_sections is not None:
summary = canonical_sections["summary"].strip()
summary_match = re.match(r"^\[([^\]\n]+)\]\s*(.*)$", summary, flags=re.DOTALL)
summary_body = summary_match.group(2).strip() if summary_match else summary
actual_task_types = [item.strip() for item in summary_match.group(1).split("+")] if summary_match else []
expected_task_types = _minimax_h3_ref_summary_task_types(user_prompt, reference_manifest)
if actual_task_types != expected_task_types:
canonical_sections["summary"] = f"[{' + '.join(expected_task_types)}] {summary_body}".rstrip()
_repair_minimax_h3_ref_single_picture_subject_definition(canonical_sections)
_repair_minimax_h3_ref_manifest_definitions(
canonical_sections,
reference_manifest,
)
_repair_minimax_h3_ref_retention_rows(
canonical_sections,
reference_manifest=reference_manifest,
)
_repair_minimax_h3_ref_subject_timeline_labels(canonical_sections)
_repair_minimax_h3_single_subject_timeline_label(canonical_sections)
canonical_sections["detailed_description"] = _repair_minimax_h3_ref_detailed_timeline(
canonical_sections["detailed_description"],
timeline_duration,
)
repaired_prompt = _rebuild_minimax_h3_ref_prompt(canonical_sections)
return repaired_prompt if repaired_prompt != prompt else prompt
heading_pattern = re.compile(
r"(?im)^[ \t]*(subject[ _-]*definitions|summary(?:[ \t]*\[[^\]\n]*\])?|retention[ _-]*analysis|"
r"detail(?:ed)?[ _-]*description|overall[ _-]*soundscape|non[ _-]*diegetic[ _-]*music)"
r"[ \t]*(?P:[ \t]*|(?=\r?$))"
)
matches = list(heading_pattern.finditer(prompt))
if not matches:
return prompt
heading_names = [
re.sub(
r"[\s-]+",
"_",
re.sub(r"\s*\[[^\]\n]*\]\s*$", "", match.group(1)).strip().casefold(),
)
for match in matches
]
heading_names = ["detailed_description" if name == "detail_description" else name for name in heading_names]
prefix = prompt[: matches[0].start()].strip()
has_colonless_heading = any(not match.group("delimiter").startswith(":") for match in matches)
if has_colonless_heading and not (
(not prefix and heading_names == list(_MINIMAX_H3_REF_FIELDS))
or (
prefix
and heading_names
== ["retention_analysis", "detailed_description", "overall_soundscape", "non_diegetic_music"]
)
):
return prompt
if prefix and "subject_definitions" not in heading_names and "summary" not in heading_names:
if heading_names not in (
["retention_analysis", "detailed_description", "overall_soundscape", "non_diegetic_music"],
["overall_soundscape", "non_diegetic_music"],
):
return prompt
if (
not any(name.startswith("summary") for name in heading_names)
and "subject_definitions" not in heading_names
and not _minimax_h3_ref_definition_items(prefix)
):
return prompt
sections = {field_name: "" for field_name in _MINIMAX_H3_REF_FIELDS}
for index, match in enumerate(matches):
raw_name = re.sub(r"\s*\[[^\]\n]*\]\s*$", "", match.group(1))
field_name = re.sub(r"[\s-]+", "_", raw_name.strip().casefold())
if field_name == "detail_description":
field_name = "detailed_description"
value_end = matches[index + 1].start() if index + 1 < len(matches) else len(prompt)
value = prompt[match.end() : value_end].strip()
if field_name in sections:
sections[field_name] = "\n".join(item for item in (sections[field_name], value) if item).strip()
if has_colonless_heading and any(not sections[field_name] for field_name in heading_names):
return prompt
prefix_recovered = False
if prefix and heading_names in (
["retention_analysis", "detailed_description", "overall_soundscape", "non_diegetic_music"],
["overall_soundscape", "non_diegetic_music"],
):
prefix_blocks = [block.strip() for block in re.split(r"\n\s*\n", prefix) if block.strip()]
retention_markers = _MINIMAX_H3_REF_VISUAL_RETENTION | _MINIMAX_H3_REF_AUDIO_RETENTION
retention_pattern = re.compile(
r"(?im)^\s*<\s*(?:Subject|Picture|Video|Audio)\s+\d+\s*>"
rf"(?:\s*(?:\([^\n)]*\)|\[[^\n\]]*\]))*\s*:\s*(?:{'|'.join(sorted(retention_markers))})\s*-\s*\S"
)
def is_retention_block(block: str) -> bool:
tagged_items = _minimax_h3_ref_definition_items(block)
lines = [line for line in block.splitlines() if line.strip()]
return bool(tagged_items) and len(retention_pattern.findall(block)) == len(tagged_items) == len(lines)
def is_definition_block(block: str) -> bool:
lines = [line for line in block.splitlines() if line.strip()]
return bool(lines) and len(_minimax_h3_ref_definition_items(block)) == len(lines)
if prefix and heading_names == ["retention_analysis", "detailed_description", "overall_soundscape", "non_diegetic_music"]:
summary_index = next(
(
index
for index, block in enumerate(prefix_blocks)
if re.match(r"^\[(?!Shot\b)[^\]\n]+\]\s+\S", block, flags=re.IGNORECASE | re.DOTALL)
),
None,
)
definition_blocks = prefix_blocks[:summary_index] if summary_index is not None else []
if (
summary_index
and summary_index == len(prefix_blocks) - 1
and all(is_definition_block(block) and not retention_pattern.search(block) for block in definition_blocks)
and is_retention_block(sections["retention_analysis"])
and re.search(r"\[Shot 1\]", sections["detailed_description"], flags=re.IGNORECASE)
and sections["overall_soundscape"]
and sections["non_diegetic_music"]
):
sections["subject_definitions"] = "\n".join(definition_blocks)
sections["summary"] = prefix_blocks[summary_index]
prefix_recovered = True
else:
return prompt
if prefix and heading_names == ["overall_soundscape", "non_diegetic_music"]:
summary_index = next(
(
index
for index, block in enumerate(prefix_blocks)
if re.match(r"^\[(?!Shot\b)[^\]\n]+\]\s+\S", block, flags=re.IGNORECASE | re.DOTALL)
),
None,
)
definition_blocks = prefix_blocks[:summary_index] if summary_index is not None else []
retention_start = summary_index + 1 if summary_index is not None else 0
retention_end = retention_start
while retention_end < len(prefix_blocks) and is_retention_block(prefix_blocks[retention_end]):
retention_end += 1
detailed_blocks = prefix_blocks[retention_end:]
detailed_shot_index = next(
(
index
for index, block in enumerate(detailed_blocks)
if re.search(r"\[Shot 1\]", block, flags=re.IGNORECASE)
),
None,
)
stray_tagged_detail = bool(
detailed_shot_index is not None
and any(_minimax_h3_ref_definition_items(block) for block in detailed_blocks[:detailed_shot_index])
)
if (
summary_index
and all(is_definition_block(block) and not retention_pattern.search(block) for block in definition_blocks)
and retention_end > retention_start
and detailed_blocks
and not stray_tagged_detail
and re.search(r"\[Shot 1\]", "\n\n".join(detailed_blocks), flags=re.IGNORECASE)
):
sections["subject_definitions"] = "\n".join(definition_blocks)
sections["summary"] = prefix_blocks[summary_index]
sections["retention_analysis"] = "\n".join(prefix_blocks[retention_start:retention_end])
sections["detailed_description"] = "\n\n".join(detailed_blocks)
prefix_recovered = True
else:
return prompt
if prefix and not prefix_recovered:
if not sections["subject_definitions"]:
sections["subject_definitions"] = prefix
elif _minimax_h3_ref_definition_items(prefix):
sections["subject_definitions"] = f"{prefix}\n{sections['subject_definitions']}".strip()
_repair_minimax_h3_ref_misplaced_detailed_description(sections)
_repair_minimax_h3_ref_single_picture_subject_definition(sections)
_repair_minimax_h3_ref_manifest_definitions(sections, reference_manifest)
definition_items: list[tuple[str, str]] = []
for tag, description in _minimax_h3_ref_definition_items(sections["subject_definitions"]):
if tag not in {item_tag for item_tag, _item_description in definition_items}:
definition_items.append((tag, description.strip(" []")))
definition_text = "\n".join(f"{tag}: {description}" for tag, description in definition_items)
if definition_items:
sections["subject_definitions"] = definition_text
summary = sections["summary"].strip()
summary_match = re.match(r"^\[([^\]\n]+)\]\s*(.*)$", summary, flags=re.DOTALL)
if summary_match:
summary = summary_match.group(2).strip()
task_prefix = " + ".join(_minimax_h3_ref_summary_task_types(user_prompt, reference_manifest))
sections["summary"] = f"[{task_prefix}] {summary}".rstrip()
_repair_minimax_h3_ref_retention_rows(
sections,
definition_items=definition_items,
reference_manifest=reference_manifest,
)
_repair_minimax_h3_ref_subject_timeline_labels(sections)
_repair_minimax_h3_single_subject_timeline_label(sections)
sections["detailed_description"] = _repair_minimax_h3_ref_detailed_timeline(
sections["detailed_description"],
timeline_duration,
)
return _rebuild_minimax_h3_ref_prompt(sections)
def _extract_json_object(text: str) -> dict[str, Any] | None:
if not text:
return None
payload, _had_boundary = _extract_delimited_final_payload(text)
cleaned = _strip_markdown(_strip_thinking(payload))
candidates = [cleaned]
candidates.append(re.sub(r"^\s*\{\s*(?=\{)", "", cleaned).strip())
first = cleaned.find("{")
last = cleaned.rfind("}")
if first >= 0 and last > first:
candidates.append(cleaned[first : last + 1])
candidates.append(re.sub(r"^\s*\{\s*(?=\{)", "", cleaned[first : last + 1]).strip())
for candidate in candidates:
try:
data = json.loads(candidate)
if isinstance(data, dict):
return data
except Exception:
continue
return None
class _DuplicatePromptPacketKey(ValueError):
pass
def _reject_duplicate_prompt_packet_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
result: dict[str, Any] = {}
for key, value in pairs:
if key in result:
raise _DuplicatePromptPacketKey(f"duplicate JSON key: {key}")
result[key] = value
return result
def _strict_prompt_packet_json(text: str) -> dict[str, Any] | None:
try:
value = json.loads(
text,
object_pairs_hook=_reject_duplicate_prompt_packet_keys,
parse_constant=lambda value: (_ for _ in ()).throw(
ValueError(f"non-finite JSON number: {value}")
),
)
except _DuplicatePromptPacketKey:
raise
except Exception:
return None
return value if isinstance(value, dict) else None
def _remove_json_trailing_commas(text: str) -> tuple[str, int]:
output: list[str] = []
in_string = False
escaped = False
removed = 0
index = 0
while index < len(text):
character = text[index]
if in_string:
output.append(character)
if escaped:
escaped = False
elif character == "\\":
escaped = True
elif character == '"':
in_string = False
index += 1
continue
if character == '"':
in_string = True
output.append(character)
index += 1
continue
if character == ",":
lookahead = index + 1
while lookahead < len(text) and text[lookahead].isspace():
lookahead += 1
if lookahead < len(text) and text[lookahead] in "}]":
removed += 1
index += 1
continue
output.append(character)
index += 1
return "".join(output), removed
def _append_missing_root_object_closer(text: str) -> str | None:
stack: list[str] = []
in_string = False
escaped = False
for character in text:
if in_string:
if escaped:
escaped = False
elif character == "\\":
escaped = True
elif character == '"':
in_string = False
continue
if character == '"':
in_string = True
elif character in "[{":
stack.append(character)
elif character in "]}":
if not stack:
return None
opening = stack.pop()
if (opening, character) not in {("[", "]"), ("{", "}")}:
return None
if in_string or escaped or stack != ["{"]:
return None
return text + "}"
def _deterministically_repaired_prompt_packet_json(
raw_output: str,
) -> tuple[dict[str, Any] | None, list[str], bool]:
"""Parse a compiler packet with a narrow, value-preserving repair set.
Grounding ledgers never use this parser. Duplicate keys are always rejected,
and no claim, prompt section, provenance list, tag ordinal, or role is
synthesized here.
"""
original = str(raw_output or "")
text = original.strip()
repairs: list[str] = []
if text.startswith("\ufeff"):
text = text.lstrip("\ufeff")
repairs.append("removed_utf8_bom")
try:
exact = _strict_prompt_packet_json(text)
except _DuplicatePromptPacketKey:
return None, [], False
if exact is not None:
return exact, repairs, not repairs
# Native thinking/final markers and one outer JSON fence are transport
# wrappers; removing them cannot change values inside the JSON object.
stripped_thinking = text
thinking_patterns = (
re.compile(r"\A.*?\s*", flags=re.IGNORECASE | re.DOTALL),
re.compile(
r"\A<\|thought\|>.*?<\|end_thought\|>\s*",
flags=re.IGNORECASE | re.DOTALL,
),
)
for pattern in thinking_patterns:
while True:
match = pattern.match(stripped_thinking)
if not match:
break
stripped_thinking = stripped_thinking[match.end() :].lstrip()
repairs.append("removed_native_thinking_wrapper")
text = stripped_thinking
delimited, had_boundary = _extract_delimited_final_payload(text)
if had_boundary:
text = delimited.strip()
repairs.append("removed_final_answer_wrapper")
fence_match = re.fullmatch(
r"\s*```(?:json)?\s*(.*?)\s*```\s*",
text,
flags=re.IGNORECASE | re.DOTALL,
)
if fence_match:
text = fence_match.group(1).strip()
repairs.append("removed_outer_json_fence")
prefix_match = re.fullmatch(
r"\s*(?:here\s+is\s+(?:the\s+)?(?:final\s+)?json|final\s+json)\s*:\s*(\{.*\})\s*",
text,
flags=re.IGNORECASE | re.DOTALL,
)
if prefix_match:
text = prefix_match.group(1).strip()
repairs.append("removed_known_json_prefix")
candidates: list[tuple[str, list[str]]] = [(text, list(repairs))]
if text.startswith("{{"):
candidates.append((text[1:].lstrip(), [*repairs, "removed_extra_outer_open_brace"]))
if text.endswith(","):
candidates.append((text[:-1].rstrip(), [*repairs, "removed_root_trailing_comma"]))
expanded: list[tuple[str, list[str]]] = []
for candidate, candidate_repairs in candidates:
expanded.append((candidate, candidate_repairs))
without_commas, removed_count = _remove_json_trailing_commas(candidate)
if removed_count:
expanded.append(
(
without_commas,
[*candidate_repairs, f"removed_trailing_commas:{removed_count}"],
)
)
closed = _append_missing_root_object_closer(candidate)
if closed is not None:
expanded.append((closed, [*candidate_repairs, "appended_missing_root_closer"]))
closed_without_commas, closed_removed_count = _remove_json_trailing_commas(
closed
)
if closed_removed_count:
expanded.append(
(
closed_without_commas,
[
*candidate_repairs,
"appended_missing_root_closer",
f"removed_trailing_commas:{closed_removed_count}",
],
)
)
if removed_count:
closed_without_commas = _append_missing_root_object_closer(without_commas)
if closed_without_commas is not None:
expanded.append(
(
closed_without_commas,
[
*candidate_repairs,
f"removed_trailing_commas:{removed_count}",
"appended_missing_root_closer",
],
)
)
seen: set[str] = set()
for candidate, candidate_repairs in expanded:
if candidate in seen:
continue
seen.add(candidate)
try:
parsed = _strict_prompt_packet_json(candidate)
except _DuplicatePromptPacketKey:
return None, [], False
if parsed is not None:
return parsed, list(dict.fromkeys(candidate_repairs)), False
return None, [], False
def _repair_single_h3_dialogue_closer_json_escape(
raw_output: str,
minimax_h3_mode: str,
) -> tuple[dict[str, Any] | None, list[str]]:
"""Remove only an invalid JSON backslash before a native ```` tag.
The recovery is limited to the model-facing one-key H3 envelope. The
resulting text must parse as strict JSON with no duplicate or extra packet
keys, and its value must already contain a complete canonical H3 prompt.
No prompt prose or dialogue delimiter is synthesized.
"""
text = str(raw_output or "").strip()
if not text:
return None, []
payload, had_boundary = _extract_delimited_final_payload(text)
if had_boundary:
text = payload.strip()
if not re.match(
r'\A\s*\{\s*"minimax_h3_prompt"\s*:',
text,
flags=re.DOTALL,
):
return None, []
packet_key_pattern = re.compile(
r'"(?:ltx_prompt|ideogram_prompt|minimax_h3_prompt|negative_prompt|scene_segments|metadata|grounding_ledger)"\s*:',
flags=re.IGNORECASE,
)
packet_keys = packet_key_pattern.findall(text)
if (
len(packet_keys) != 1
or not re.fullmatch(
r'"minimax_h3_prompt"\s*:',
packet_keys[0],
flags=re.IGNORECASE,
)
):
return None, []
repaired_text, repair_count = re.subn(
r"\\(?=)",
"",
text,
flags=re.IGNORECASE,
)
if repair_count <= 0:
return None, []
try:
parsed = _strict_prompt_packet_json(repaired_text)
except _DuplicatePromptPacketKey:
return None, []
if (
parsed is None
or set(parsed) != {"minimax_h3_prompt"}
or not isinstance(parsed.get("minimax_h3_prompt"), str)
or not _extract_complete_minimax_h3_plain_prompt(
parsed["minimax_h3_prompt"],
minimax_h3_mode,
)
):
return None, []
repairs = []
if had_boundary:
repairs.append("removed_final_answer_wrapper")
repairs.append(f"removed_invalid_h3_dialogue_closer_json_escape:{repair_count}")
return parsed, repairs
def _single_paragraph(text: str) -> str:
return _strip_generation_artifacts(re.sub(r"\s+", " ", (text or "").replace("\n", " ")).strip())
def _ensure_terminal_period(text: str) -> str:
value = _single_paragraph(text)
if not value:
return ""
if value[-1] in ".!?":
return value
return f"{value}."
def _sentence_case(text: str) -> str:
value = _single_paragraph(text)
if value and value[0].islower():
return value[0].upper() + value[1:]
return value
def _clean_prompt_request_text(text: str) -> str:
value = _single_paragraph(text)
if _is_media_reference_only_request(value):
return ""
patterns = [
r"^(?:create|write|generate|make)\s+(?:an?\s+)?(?:ltx(?:[- ]?2(?:\.(?:3|5))?)?\s+)?(?:script|prompt)\s+for\s+(?:image|video)\s+generation\s+(?:featuring|with|of|about)\s+",
r"^(?:create|write|generate|make)\s+(?:an?\s+)?(?:ltx(?:[- ]?2(?:\.(?:3|5))?)?\s+)?(?:script|prompt)\s+(?:featuring|with|of|about)\s+",
r"^(?:create|write|generate|make)\s+(?:an?\s+)?(?:image|video)\s+(?:generation\s+)?(?:script|prompt)\s+(?:featuring|with|of|about)\s+",
]
cleaned = value
for pattern in patterns:
cleaned = re.sub(pattern, "", cleaned, flags=re.IGNORECASE).strip()
cleaned = re.sub(r"\.{2,}", ".", cleaned)
cleaned = re.sub(r"\s+([,.!?])", r"\1", cleaned).strip()
return _normalize_hex_text(_sentence_case(cleaned or value))
def _clean_visual_description(text: str, max_chars: int = 3000) -> str:
value = _strip_generation_artifacts(_strip_markdown(_strip_thinking(str(text or ""))))
value = re.sub(r"\s+", " ", value).strip()
if max_chars > 0 and len(value) > max_chars:
value = value[:max_chars].rsplit(" ", 1)[0].strip() or value[:max_chars].strip()
return value
def _visual_description_from_metadata(media_metadata: dict[str, Any] | None) -> str:
if not isinstance(media_metadata, dict):
return ""
return _clean_visual_description(str(media_metadata.get("visual_description", "")))
def _normalize_media_synthesis_mode(value: Any) -> str:
text = str(value or "video_recreation").strip().lower().replace("-", "_").replace(" ", "_")
if text in {"image_identity_video_control", "identity_image_video_control", "image_identity", "identity_reference"}:
return "image_identity_video_control"
return "video_recreation"
def _is_image_identity_video_control(media_metadata: dict[str, Any] | None) -> bool:
return _normalize_media_synthesis_mode(
media_metadata.get("media_synthesis_mode") if isinstance(media_metadata, dict) else ""
) == "image_identity_video_control"
def _is_minimax_h3_reference_context(media_metadata: dict[str, Any] | None) -> bool:
if not isinstance(media_metadata, dict):
return False
value = str(media_metadata.get("minimax_h3_mode", "")).strip().lower().replace("-", "_")
return value in {"ref2va", "ref2v", "r2v", "reference", "reference_to_video", "reference_to_audio_video"}
def _attached_reference_image_count(media_metadata: dict[str, Any] | None) -> int:
"""Return only reference images that are present in the processor batch.
Older image+video contexts predate the explicit attachment flag, so their
declared count remains authoritative. Newer contexts fail closed when the
sampler records that a declared still could not be attached.
"""
if not isinstance(media_metadata, dict):
return 0
source = str(media_metadata.get("source", "none") or "none").strip().lower()
if _is_minimax_h3_reference_context(media_metadata):
declared = int(
max(
0.0,
_safe_float(
media_metadata.get(
"minimax_h3_reference_image_batch_count",
media_metadata.get("reference_image_count", 0),
),
0.0,
),
)
)
else:
declared = int(max(0.0, _safe_float(media_metadata.get("reference_image_count"), 0.0)))
if "reference_image_backend_attached" in media_metadata:
if not bool(media_metadata.get("reference_image_backend_attached")):
return 0
elif source not in {"image", "image+video"}:
return 0
if source == "image+video" and declared == 0 and not _is_minimax_h3_reference_context(media_metadata):
# Compatibility for legacy ordinary image+video contexts.
return 1
return declared
def _synthesis_contract_text(media_metadata: dict[str, Any] | None) -> str:
if _is_image_identity_video_control(media_metadata):
return (
"Use the reference image as the identity and appearance source for the output subject, including face, hair, body type, wardrobe, accessories, styling, and distinguishing visual traits. "
"Use the video frames only as control structure for pose, depth, canny/edge layout, composition, action, timing, camera motion, blocking, and scene geometry. "
"Do not copy the control-video subject's clothing, hair, face, body type, accessories, or style unless the user explicitly asks for those video-subject traits."
)
return "Recreate the source video or media evidence directly unless the user prompt asks for a different transformation."
def _copy_synthesis_metadata(target: dict[str, Any], media_metadata: dict[str, Any]) -> None:
for key in (
"media_synthesis_mode",
"image_role",
"video_role",
"reference_image_count",
"reference_image_backend_attached",
"reference_image_resized_for_video_control",
"reference_image_resize_mode",
"reference_image_original_width",
"reference_image_original_height",
"reference_image_backend_width",
"reference_image_backend_height",
"video_sampled_frame_count",
"synthesis_contract",
"ltx25_context_schema",
"ltx_generation_mode_hint",
"ltx_first_frame_attached",
"ltx_last_frame_attached",
"ltx_frame_pair_attached",
"ltx_frame_roles",
"image_roles",
"ltx_first_frame_width",
"ltx_first_frame_height",
"ltx_last_frame_width",
"ltx_last_frame_height",
):
if key in media_metadata:
target[key] = media_metadata[key]
def _media_source_has_pixels(media_metadata: dict[str, Any] | None) -> bool:
if not isinstance(media_metadata, dict):
return False
source = str(media_metadata.get("source", "none")).lower()
return source in {"image", "video", "image+video"} or bool(media_metadata.get("sampled_frame_count"))
def _media_pixels_sent_to_backend(media_metadata: dict[str, Any] | None) -> bool:
return bool(isinstance(media_metadata, dict) and media_metadata.get("pixels_sent_to_backend"))
def _media_grounding_mode(media_metadata: dict[str, Any] | None) -> str:
if isinstance(media_metadata, dict) and isinstance(media_metadata.get("verified_grounding_ledger"), dict):
return "verified_ledger"
if not _media_source_has_pixels(media_metadata):
return "none"
if _media_pixels_sent_to_backend(media_metadata):
return "pixels"
if _visual_description_from_metadata(media_metadata):
return "visual_description"
return "metadata_only"
def _video_transport_mode() -> str:
value = os.environ.get("DG_VIDEO_TRANSPORT", "sampled_frame_images").strip().lower()
value = value.replace("-", "_").replace(" ", "_")
if value in {"video", "video_token", "video_tokens", "native_video"}:
return "video_tokens"
return "sampled_frame_images"
def _video_frame_timecodes(media_metadata: dict[str, Any] | None, frame_count: int) -> list[float]:
if frame_count <= 0:
return []
if not isinstance(media_metadata, dict):
return [float(i) for i in range(frame_count)]
indices = media_metadata.get("sampled_indices")
source_fps = _safe_float(media_metadata.get("source_fps"), 0.0)
if isinstance(indices, list) and source_fps > 0:
values: list[float] = []
for index in indices[:frame_count]:
try:
values.append(float(index) / source_fps)
except Exception:
values.append(float(len(values)))
if len(values) == frame_count:
return values
duration = _safe_float(media_metadata.get("duration_seconds"), 0.0)
if duration > 0 and frame_count > 1:
step = duration / max(1, frame_count - 1)
return [idx * step for idx in range(frame_count)]
sample_fps = max(0.001, _safe_float(media_metadata.get("sample_fps"), 1.0))
return [idx / sample_fps for idx in range(frame_count)]
def _display_video_frame_count(media_metadata: dict[str, Any] | None, frame_count: int) -> int:
if not isinstance(media_metadata, dict):
return frame_count
video_count = int(max(0, _safe_float(media_metadata.get("video_sampled_frame_count"), 0.0)))
if video_count > 0:
return min(frame_count, video_count)
return frame_count
def _sampled_video_frame_sequence_intro(media_metadata: dict[str, Any] | None, frame_count: int) -> str:
reference_count = min(frame_count, _attached_reference_image_count(media_metadata))
video_frame_count = _display_video_frame_count(media_metadata, max(0, frame_count - reference_count))
timecodes = _video_frame_timecodes(media_metadata, video_frame_count)
shown = ", ".join(f"{idx + 1}={seconds:.2f}s" for idx, seconds in enumerate(timecodes[:24]))
if video_frame_count > 24:
shown = f"{shown}, ..."
source = str(media_metadata.get("source", "video")) if isinstance(media_metadata, dict) else "video"
if _is_minimax_h3_reference_context(media_metadata):
picture_intro = (
f"The first {reference_count} attached image(s) are ordered MiniMax H3 reference pictures through . "
if reference_count
else "No reference pictures are attached for pixel analysis. "
)
return (
f"{picture_intro}The remaining images are ordered sampled frames from