"""Disk-backed recursive clip chains specialized for MiniMax H3. The visible graph contains one H3 sampling body. Chain Start and Chain End recursively clone that body with ComfyUI's GraphBuilder, carrying only the previous clip's context tail and compact AV latent into the next iteration. Each iteration is persisted before recursion, so a long chain can resume from the first unfinished clip instead of starting over. The recursive graph traversal is adapted from Ethanfel's SxCP loop nodes in ComfyUI-Prompt-Builder, using the same ComfyUI expansion pattern with a single, typed MiniMax chain state rather than arbitrary carry sockets. """ from __future__ import annotations import asyncio import concurrent.futures import gc import hashlib import ipaddress import itertools import json import logging import math import os import re import secrets import shutil import subprocess import struct import sys import time import uuid import wave from collections import deque from contextlib import contextmanager, nullcontext from datetime import datetime, timezone from fractions import Fraction from typing import Any import folder_paths try: import av except ImportError: # ComfyUI normally ships PyAV. av = None try: import torch except ImportError: # ComfyUI always ships torch; keeps local imports clear. torch = None try: import numpy as np except ImportError: # NumPy ships with ComfyUI and PyAV. np = None try: from PIL import Image, ImageOps, PngImagePlugin except ImportError: # Pillow ships with ComfyUI. Image = ImageOps = PngImagePlugin = None try: from safetensors.torch import load_file as _st_load, save_file as _st_save except ImportError: _st_load = _st_save = None try: from comfy_execution.graph_utils import GraphBuilder, ExecutionBlocker, is_link except ImportError: GraphBuilder = None ExecutionBlocker = None def is_link(value): return isinstance(value, list) and len(value) == 2 try: from aiohttp import web from server import PromptServer except ImportError: web = None PromptServer = None from .nodes import ( MiniMaxH3MotionContext, _claim_inline_patch_ownership, _prepare_native_guide_conditioning, _resize, _streams_from_latent, ) from .motion_context_upstream import apply_motion_context from .chapter_resolution import ( normalize_resolution, scene_resolution, saved_resolution, apply_chapter_resolutions, common_saved_resolution, ) from .prompt_history import PromptHistoryStore from .prompt_optimizer import optimize_prompt_payload from .run_manager import RunArchiveManager, archive_policy_inputs from .asset_store import MAX_DIRECT_ASSET_BINDINGS, RunAssetStore from .reference_cache_store import ( FORMAT as REFERENCE_CACHE_FORMAT, ReferenceTensorStore, objects_digest, ) from .reference_cache_migration import ReferenceCacheMigrator, matches_legacy from .reference_cache_usage import note_converted_use, confirm_saved_use from .project_assets import ( PROJECT_ASSET_FORMAT, ProjectAssetStore, ) from .av_timing import ( AUDIO_TRIM_FRAMES_KEY, AUDIO_WITH_OVERLAP_FRAMES_KEY, AUDIO_WITH_OVERLAP_WAVEFORM_KEY, conform_waveform_length, sample_boundary_from_frames, ) from .contracts_v05 import ( ADVANCED_TRANSITION_PRESETS, AV_TRANSITION_CONTEXT_LENGTHS, AUDIO_POLICY_VERSION, CHAIN_POLICY_VERSION, CONTEXT_SPATIAL_PROXY_MODES, CONTEXT_SPATIAL_PROXY_RECIPE, CONTINUATION_POLICIES, DETAIL_AV_RECIPE, DRIFT_CONTROL_AV_RECIPE, LATENT_COLOR_CARRY_RECIPE, DEPENDENCY_SCOPES, FINAL_AUDIO_POLICIES, GENERATION_AUDIO_PROFILES, GENERATION_SCENE_PROFILES, GENERATED_CONTINUITY_POLICIES, LIP_SYNC_OPTIONS_VERSION, PRIMARY_TRANSITION_PRESETS, PREFLIGHT_VERSION, SCENE_DEPENDENCY_VERSION, SOURCE_REFERENCE_POLICIES, SOURCE_TIMELINE_VERSION, TRANSITION_CONTEXT_LENGTHS, TRANSITION_POLICY_VERSION, audio_policy as _contract_audio_policy, chain_policy as _contract_chain_policy, compose_chain_policy as _contract_compose_chain_policy, migrate_continuation_mode, migrate_legacy_audio_mode, masked_song_options as _contract_masked_song_options, paired_audio_policy as _contract_paired_audio_policy, transition_policy as _contract_transition_policy, ) from .checkpoint_manager import ( CheckpointDeleteBlocked, CheckpointGraphManager, checkpoint_audio_context_length, checkpoint_revision_token, checkpoint_run_lock, ) _LOG = logging.getLogger("minimax_h3_context_loop.chain") FPS = 24 AUDIO_HZ = 40 PLAN_VERSION = 2 MAX_SHOTS = 128 MAX_SEED = 0xFFFFFFFFFFFFFFFF MAX_REVIEW_CANDIDATES = 20 MAX_H3_FRAMES = 3592 # largest 17k+5 value accepted by H3's 3600-frame socket PNG_EXPORT_AUTO_WORKERS_CAP = 8 PNG_EXPORT_MIN_CHUNK_FRAMES = 8 PNG_EXPORT_MAX_CHUNK_FRAMES = 64 PNG_EXPORT_HASH_CACHE_FORMAT = "h3_png_checkpoint_hash_cache_v1" CHAPTER_MANIFEST_FORMAT = "h3_chain_chapter_manifest_v1" BOUNDARY_TONE_MATCH_MODES = ("off", "auto") BOUNDARY_TONE_MATCH_CONTEXT_FRAMES = 8 BOUNDARY_TONE_MATCH_SEARCH_FRAMES = 4 BOUNDARY_TONE_MATCH_MIN_JUMP = 1.25 BOUNDARY_TONE_MATCH_MAX_SHIFT = 6.0 ASSEMBLY_COLOR_STABILIZATION_MODES = ("off", "scene_1_anchor") ASSEMBLY_COLOR_STABILIZATION_RAMP_FRAMES = 72 ASSEMBLY_COLOR_STABILIZATION_STRENGTH = 0.5 ASSEMBLY_COLOR_STABILIZATION_MAX_LUMA_SHIFT = 6.0 ASSEMBLY_COLOR_STABILIZATION_MAX_SATURATION_CHANGE = 0.06 MASKED_AUDIO_CONTRACT = "raw_source_window_v2" PLAN_STUDIO_PREVIEW_TTL_SECONDS = 6 * 60 * 60 PLAN_STUDIO_PREVIEW_MAX_SESSIONS = 32 PLAN_STUDIO_PREVIEW_MAX_HEIGHT = 480 PLAN_STUDIO_WAVEFORM_SAMPLE_RATE = 192 PLAN_STUDIO_WAVEFORM_POINTS_PER_SECOND = 12 PLAN_STUDIO_THUMBNAIL_WIDTH = 320 _PLAN_STUDIO_SOURCE_PREVIEWS: dict[str, dict[str, Any]] = {} _PLAN_STUDIO_PREVIEW_BUILD_LOOP: asyncio.AbstractEventLoop | None = None _PLAN_STUDIO_PREVIEW_BUILD_GATE: asyncio.Semaphore | None = None _PLAN_STUDIO_PREVIEW_BUILD_TASKS: dict[str, asyncio.Task[str]] = {} _PLAN_STUDIO_THUMBNAIL_BUILD_LOOP: asyncio.AbstractEventLoop | None = None _PLAN_STUDIO_THUMBNAIL_BUILD_GATE: asyncio.Semaphore | None = None _PLAN_STUDIO_THUMBNAIL_BUILD_TASKS: dict[str, asyncio.Task[str]] = {} H3_CONTEXT_LENGTHS = ( 1, 5, 22, 39, 56, 73, 90, 107, 124, 141, 158, 175, 192, 209, 226, 243, ) AUDIO_MODES = ("source_track", "generated_audio", "source_plus_timeline") CONTINUATION_MODES = ( "guide", "tone_carry_guide", "latent_guide", "tapered_guide", "masked_av", "tapered_av", "feathered_av", "audio_feathered_av", "drift_control_av", "color_stable_drift_av") SCENE_LORA_ROUTES = ( "base", *(chr(ord("a") + offset) for offset in range(26))) LOOP_MEMORY_POLICIES = ("off", "unload_models", "fresh_scene") GUIDE_CONTINUATION_MODES = frozenset(( "guide", "tone_carry_guide", "latent_guide", "tapered_guide")) MASKED_CONTINUATION_MODES = frozenset(( "masked_av", "tapered_av", "feathered_av", "audio_feathered_av", "drift_control_av", "color_stable_drift_av")) DISPOSABLE_PREFIX_CONTINUATION_MODES = frozenset(( "tapered_av", "drift_control_av", "color_stable_drift_av")) DRIFT_CONTROL_CONTINUATION_MODES = frozenset(( "drift_control_av", "color_stable_drift_av")) LATENT_COLOR_CARRY_CONTINUATION_MODES = frozenset(( "color_stable_drift_av",)) RGB_PROXY_GUIDE_MODES = frozenset(( "guide", "tone_carry_guide", "tapered_guide")) REFERENCE_AUDIO_TIMELINE_MODES = ("standalone", "source_timeline") MOTION_REFERENCE_SHORT_EDGES = ("384", "512", "768", "source") _LAZY_INPUT_MISSING = object() # These Plan-wide controls describe how the next scene consumes an immutable # predecessor. Their effective values are still recorded on generated scenes, # but changing them must not by itself rewrite the predecessor's identity. _NEXT_SCENE_COMPATIBILITY_KEYS = frozenset(( "continuation_mode", "context_length", "context_storage_length", )) PLAN_TYPE = "H3_CHAIN_PLAN" STATE_TYPE = "H3_CHAIN_STATE" FLOW_TYPE = "H3_CHAIN_FLOW" SEGMENT_TYPE = "H3_CHAIN_SEGMENT" MANIFEST_TYPE = "H3_CHAIN_MANIFEST" EXTERNAL_CONTEXT_TYPE = "H3_CHAIN_EXTERNAL_CONTEXT" REFERENCE_SCHEDULE_TYPE = "H3_REFERENCE_SCHEDULE" TAGGED_REFERENCE_TYPE = "H3_TAGGED_REFERENCES" REFMOD_SOURCE_LIST_TYPE = "H3_REF_LIST" TAGGED_SCENE_OPTIONS_TYPE = "H3_TAGGED_SCENE_OPTIONS" SCENE_DATA_TYPE = "H3_SCENE_DATA" SEMANTIC_ANCHOR_DRAFT_TYPE = "H3_SEMANTIC_ANCHOR_DRAFT" SEMANTIC_PRESENTATION_TYPE = "H3_SEMANTIC_PRESENTATION" LAZY_MOTION_SOURCE_TYPE = "H3_LAZY_MOTION_SOURCE" SOURCE_TIMELINE_TYPE = "H3_SOURCE_TIMELINE" PROJECT_ASSETS_TYPE = "H3_PROJECT_ASSETS" AUDIO_POLICY_TYPE = "H3_AUDIO_POLICY" TRANSITION_POLICY_TYPE = "H3_TRANSITION_POLICY" CHAIN_POLICY_TYPE = "H3_CHAIN_POLICY" LIP_SYNC_OPTIONS_TYPE = "H3_LIP_SYNC_OPTIONS" PREFLIGHT_TYPE = "H3_PREFLIGHT_REPORT" REFERENCE_SCHEDULE_VERSION = 1 REFERENCE_FINGERPRINT_LINEAGE_VERSION = 1 TAGGED_SCENE_OPTIONS_VERSION = 1 SCENE_DATA_VERSION = 1 SEMANTIC_ANCHOR_REGISTRY_VERSION = 1 LAZY_MOTION_SOURCE_VERSION = 3 SEMANTIC_PRESENTATION_VERSION = 1 EXTERNAL_REFMOD_CONTRACT_VERSION = 2 _PENDING_REVIEWS: dict[str, dict[str, Any]] = {} _ACTIVE_CANDIDATE_BATCHES: dict[str, dict[str, Any]] = {} _PENDING_FINAL_REVIEW_PREVIEWS: dict[ tuple[str, str], dict[str, Any] ] = {} _FFMPEG_PROBE_CACHE: dict[str, bool] = {} def _canonical_json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) def _fingerprint(value: Any) -> str: return hashlib.sha256(_canonical_json(value).encode("utf-8")).hexdigest() def _validate_audio_policy(value: Any) -> dict[str, Any]: if not isinstance(value, dict): raise ValueError( "H3 Audio Policy must come from MiniMax H3 Audio Policy.") if value.get("version") != AUDIO_POLICY_VERSION: raise ValueError( "H3 Audio Policy version is missing or obsolete.") return _contract_audio_policy( value.get("final_audio"), value.get("source_reference"), value.get("generated_continuity"), value.get("source_audio_target", "off"), value.get("lip_sync_options")) def _resolved_audio_policy(value: Any) -> dict[str, Any]: """Resolve an explicit 0.5 policy or faithfully migrate a 0.4 mode.""" compatibility = ( value.get("compatibility") if isinstance(value, dict) and isinstance( value.get("compatibility"), dict) else value) if not isinstance(compatibility, dict): raise ValueError("H3 audio policy requires Plan compatibility data.") explicit = compatibility.get("audio_policy") if explicit is not None: return _validate_audio_policy(explicit) return migrate_legacy_audio_mode(compatibility.get("audio_mode")) def _legacy_audio_mode_for_policy(policy: Any) -> str: resolved = _validate_audio_policy(policy) for mode in AUDIO_MODES: if migrate_legacy_audio_mode(mode) == resolved: return mode return "custom" def _audio_policy_summary(value: Any) -> str: policy = _resolved_audio_policy(value) summary = "final=%s/ref=%s/carry=%s" % ( policy["final_audio"], policy["source_reference"], policy["generated_continuity"]) if policy.get("source_audio_target") == "locked": summary += "/target=locked" return summary def _optional_scene_audio_axis( shot: Any, key: str, allowed: tuple[str, ...]) -> str | None: """Return one validated scene override, or None for inherited behavior.""" if not isinstance(shot, dict) or key not in shot: return None raw = shot.get(key) if raw is None or (isinstance(raw, str) and not raw.strip()): return None value = str(raw).strip().lower() if value == "inherit": return None if value not in allowed: raise ValueError( "Unknown H3 scene %s override %r; expected inherit or one of %s." % (key.replace("_", " "), raw, allowed)) return value def _resolved_scene_audio_policy( value: Any, shot: Any = None) -> dict[str, Any]: """Overlay one scene's generation axes on the Plan-wide audio policy. Final mux choice deliberately remains Plan-wide. Locking the source target keeps the existing contract rule: it disables loose source reference and predecessor generated-audio carry for that scene. """ policy = _resolved_audio_policy(value) source_reference = _optional_scene_audio_axis( shot, "source_reference", SOURCE_REFERENCE_POLICIES) generated_continuity = _optional_scene_audio_axis( shot, "generated_continuity", GENERATED_CONTINUITY_POLICIES) source_audio_target = _optional_scene_audio_axis( shot, "source_audio_target", ("off", "locked")) return _contract_audio_policy( policy["final_audio"], source_reference or policy["source_reference"], generated_continuity or policy["generated_continuity"], source_audio_target or policy.get("source_audio_target", "off"), policy.get("lip_sync_options"), ) def _resolved_lip_sync_options( value: Any, shot: Any = None) -> dict[str, Any] | None: """Return the contextual song recipe only for an active locked target.""" policy = _resolved_scene_audio_policy(value, shot) if policy.get("source_audio_target", "off") != "locked": return None options = policy.get("lip_sync_options") return (_contract_masked_song_options(options) if options is not None else None) def _audio_policy_locks_source_audio(value: Any, shot: Any = None) -> bool: return _resolved_scene_audio_policy(value, shot).get( "source_audio_target", "off") == "locked" def _audio_source_requirements(value: Any) -> list[dict[str, Any]]: """Describe every effective setting that makes source audio mandatory.""" policy = _resolved_audio_policy(value) requirements: list[dict[str, Any]] = [] if policy["final_audio"] == "source": requirements.append({ "scope": "final output", "setting": "final_audio", "value": "source", "origin": "Chain Policy", }) shots = value.get("shots") if isinstance(value, dict) else None if isinstance(shots, list): for fallback_index, shot in enumerate(shots, 1): resolved = _resolved_scene_audio_policy(value, shot) scene = int(shot.get("index", fallback_index)) scene_id = str(shot.get("id") or "scene_%d" % scene) for setting, active_value, allowed in ( ("source_reference", "on", SOURCE_REFERENCE_POLICIES), ("source_audio_target", "locked", ("off", "locked"))): if resolved.get(setting, "off") != active_value: continue override = _optional_scene_audio_axis( shot, setting, allowed) requirements.append({ "scope": "scene", "scene": scene, "scene_id": scene_id, "setting": setting, "value": active_value, "origin": ( "scene override" if override is not None else "Chain Policy"), }) else: for setting, active_value in ( ("source_reference", "on"), ("source_audio_target", "locked")): if policy.get(setting, "off") == active_value: requirements.append({ "scope": "Plan", "setting": setting, "value": active_value, "origin": "Chain Policy", }) return requirements def _format_audio_source_requirement(requirement: dict[str, Any]) -> str: if requirement.get("scope") == "scene": location = 'Scene %d "%s"' % ( int(requirement["scene"]), str(requirement["scene_id"])) else: location = str(requirement.get("scope") or "Plan") return "%s: %s=%s (%s)" % ( location, str(requirement.get("setting") or "audio_policy"), str(requirement.get("value") or "required"), str(requirement.get("origin") or "effective policy")) def _audio_policy_requires_source(value: Any) -> bool: policy = _resolved_audio_policy(value) if policy["final_audio"] == "source": return True shots = value.get("shots") if isinstance(value, dict) else None if isinstance(shots, list): return any( _audio_policy_uses_source_reference(value, shot) or _audio_policy_locks_source_audio(value, shot) for shot in shots) return (policy["source_reference"] == "on" or policy.get("source_audio_target") == "locked") def _audio_policy_uses_source_reference(value: Any, shot: Any = None) -> bool: return _resolved_scene_audio_policy( value, shot)["source_reference"] == "on" def _audio_policy_uses_generated_continuity( value: Any, shot: Any = None) -> bool: return _resolved_scene_audio_policy( value, shot)["generated_continuity"] == "on" def _audio_policy_final(value: Any) -> str: return _resolved_audio_policy(value)["final_audio"] def _paired_audio_policy(value: Any) -> str: return _contract_paired_audio_policy(value) def _validate_chain_policy(value: Any) -> dict[str, Any]: if not isinstance(value, dict): raise ValueError( "H3 Chain Policy must come from MiniMax H3 Generation Profile, " "Manual Chain Policy, Advanced Policy, or the Legacy 0.4 " "Policy Adapter.") if value.get("version") != CHAIN_POLICY_VERSION: raise ValueError("H3 Chain Policy version is missing or obsolete.") return _contract_compose_chain_policy( value.get("audio_policy"), value.get("transition_policy"), audio_context_length=value.get("audio_context_length", 22)) def _validate_transition_policy(value: Any) -> dict[str, Any]: if not isinstance(value, dict): raise ValueError( "H3 Transition Policy must come from MiniMax H3 Transition Policy.") if value.get("version") != TRANSITION_POLICY_VERSION: raise ValueError( "H3 Transition Policy version is missing or obsolete.") return _contract_transition_policy( value.get("preset"), expert_override=bool(value.get("expert_override", False)), continuation_mode=value.get("continuation_mode"), context_length=value.get("context_length")) def _resolved_transition_policy(value: Any) -> dict[str, Any]: compatibility = ( value.get("compatibility") if isinstance(value, dict) and isinstance( value.get("compatibility"), dict) else value) if not isinstance(compatibility, dict): raise ValueError( "H3 transition policy requires Plan compatibility data.") explicit = compatibility.get("transition_policy") if explicit is not None: return _validate_transition_policy(explicit) mode = migrate_continuation_mode( compatibility.get("continuation_mode", "guide")) context = int(compatibility.get("context_length", 0)) preset = "legacy" for candidate in ( "cut", "guide", "tone_guide", "latent_guide", "detail_guide", "detail_av", "drift_av", "color_drift_av", "hard_av", "soft_av", "audio_feather_av"): resolved = _contract_transition_policy(candidate) if (resolved["continuation_mode"] == mode and int(resolved["context_length"]) == context): preset = candidate break return { "version": TRANSITION_POLICY_VERSION, "preset": preset, "continuation_mode": mode, "context_length": context, "expert_override": False, } def _transition_policy_summary(value: Any) -> str: policy = _resolved_transition_policy(value) return "%s/%s/%df" % ( policy["preset"], policy["continuation_mode"], int(policy["context_length"])) def _transition_policy_display(value: Any) -> str: """Return the human-facing preset -> implementation mapping.""" policy = _resolved_transition_policy(value) preset_labels = { "cut": "Cut", "guide": "Guide", "tone_guide": "Tone Carry Guide", "latent_guide": "Latent Guide", "detail_guide": "Detail Guide", "detail_av": "Detail AV", "drift_av": "Drift-Control AV", "color_drift_av": "Color-Stable Drift AV", "hard_av": "Hard AV", "soft_av": "Soft AV", "audio_feather_av": "Audio Feather AV (legacy alias)", "legacy": "Legacy", } implementation_labels = { "guide": "Guide", "tone_carry_guide": "Tone Carry Guide", "latent_guide": "Latent Guide", "tapered_guide": "Tapered Guide", "masked_av": "Masked AV", "tapered_av": "Tapered AV", "feathered_av": "Feathered AV", "audio_feathered_av": "Audio-Feathered AV", "drift_control_av": "Drift-Control AV", "color_stable_drift_av": "Color-Stable Drift AV", } preset = str(policy["preset"]) implementation = str(policy["continuation_mode"]) return "%s -> %s + %d frames" % ( preset_labels.get(preset, preset), implementation_labels.get(implementation, implementation), int(policy["context_length"])) def _safe_name(value: str, fallback: str = "chain") -> str: text = re.sub(r"[^A-Za-z0-9._-]+", "_", str(value or "").strip()) text = text.strip("._-") return (text or fallback)[:96] def _strict_run_name(value: Any) -> str: """Return an exact Run identity; never redirect a selected chapter.""" requested = str(value or "").strip() normalized = _safe_name(requested, "") if not normalized: raise ValueError("A non-empty H3 chain run_name is required.") if requested != normalized: raise ValueError( "H3 run_name must be 1-96 characters, start and end with a " "letter or number, and contain only letters, numbers, '.', '_' " "or '-'. Use %r for this Run." % normalized) return normalized def _expand_filename_date(value: str, now: datetime | None = None) -> str: """Expand ComfyUI-style date tokens before filename sanitization.""" current = now or datetime.now() replacements = { "yyyy": "%Y", "yy": "%y", "MM": "%m", "dd": "%d", "HH": "%H", "hh": "%I", "mm": "%M", "ss": "%S", } def replace_date(match: re.Match[str]) -> str: pattern = match.group(1) strftime_pattern = re.sub( r"yyyy|yy|MM|dd|HH|hh|mm|ss", lambda token: replacements[token.group(0)], pattern, ) return current.strftime(strftime_pattern) text = re.sub(r"%date:([^%]+)%", replace_date, str(value or "")) simple_tokens = { "%year%": "%Y", "%month%": "%m", "%day%": "%d", "%hour%": "%H", "%minute%": "%M", "%second%": "%S", } for token, pattern in simple_tokens.items(): text = text.replace(token, current.strftime(pattern)) return text def _available_versioned_path(path: str) -> str: """Return path unchanged when free, otherwise add a numeric version.""" if not os.path.exists(path): return path root, extension = os.path.splitext(path) version = 1 while True: candidate = "%s_%03d%s" % (root, version, extension) if not os.path.exists(candidate): return candidate version += 1 def _safe_output_subfolder(value: str) -> str: """Return a safe, date-expanded subfolder relative to ComfyUI output.""" text = _expand_filename_date(value).strip().replace("\\", "/") if not text or text == ".": return "" if text.startswith("/") or re.match(r"^[A-Za-z]:", text): raise ValueError( "H3 Chain Assemble output_subfolder must be relative to the " "ComfyUI output folder.") parts = [] for part in text.split("/"): if not part or part == ".": continue if part == "..": raise ValueError( "H3 Chain Assemble output_subfolder cannot contain '..'.") safe = _safe_name(part, "") if not safe: raise ValueError( "H3 Chain Assemble output_subfolder contains an invalid " "folder name.") parts.append(safe) return os.path.join(*parts) if parts else "" def _copy_final_to_output(final_path: str, output_subfolder: str) -> str: """Copy a completed final beside regular ComfyUI outputs without overwrite.""" root = os.path.abspath(_output_root()) relative_folder = _safe_output_subfolder(output_subfolder) destination_dir = os.path.abspath(os.path.join(root, relative_folder)) if os.path.commonpath([root, destination_dir]) != root: raise ValueError( "H3 Chain Assemble output_subfolder escapes the output folder.") os.makedirs(destination_dir, exist_ok=True) destination = _available_versioned_path(os.path.join( destination_dir, os.path.basename(final_path))) temporary = "%s.%s.tmp" % (destination, uuid.uuid4().hex) try: shutil.copy2(final_path, temporary) os.replace(temporary, destination) finally: _safe_unlink(temporary) return destination def _prompt_text(value: Any, label: str) -> str: """Normalize a prompt string or a human-editable JSON array of lines.""" if isinstance(value, list): if not all(isinstance(line, str) for line in value): raise ValueError("%s line arrays may contain only strings." % label) return "\n".join(value).strip() return str(value or "").strip() def _resolve_dynamic_prompt( value: str, seed: int, scope: str = "" ) -> tuple[str, bool]: """Resolve Comfy-style ``{one|two}`` groups deterministically. Resolution is deliberately backend- and scene-local. Expanding the complete Plan JSON through ComfyUI's frontend dynamic-prompt serializer would rerandomize unrelated scenes and make checkpoint recovery depend on queue timing. The same dedicated prompt seed, scope, and template always produce the same effective prompt, independently of the sampler seed. Nested groups and the escapes ``\\{``, ``\\}``, ``\\|``, and ``\\\\`` are supported. Braces without an unescaped top-level pipe remain literal. """ text = str(value or "") seed = int(seed) if seed < 0 or seed > MAX_SEED: raise ValueError("Dynamic prompt seed is outside the uint64 range.") group_number = 0 max_groups = 256 max_depth = 32 def escaped_character(source: str, position: int) -> tuple[str, int] | None: if (source[position] == "\\" and position + 1 < len(source) and source[position + 1] in "{}|\\"): return source[position + 1], position + 2 return None def matching_brace(source: str, opening: int) -> int | None: depth = 0 position = opening while position < len(source): escaped = escaped_character(source, position) if escaped is not None: position = escaped[1] continue character = source[position] if character == "{": depth += 1 elif character == "}": depth -= 1 if depth == 0: return position position += 1 return None def alternatives(source: str) -> list[str] | None: parts = [] start = 0 depth = 0 position = 0 while position < len(source): escaped = escaped_character(source, position) if escaped is not None: position = escaped[1] continue character = source[position] if character == "{": depth += 1 elif character == "}" and depth: depth -= 1 elif character == "|" and depth == 0: parts.append(source[start:position]) start = position + 1 position += 1 if not parts: return None parts.append(source[start:]) return parts def render(source: str, depth: int = 0) -> tuple[str, bool]: nonlocal group_number if depth > max_depth: raise ValueError( "Dynamic prompt nesting exceeds the maximum depth of %d." % max_depth) output = [] used = False position = 0 while position < len(source): escaped = escaped_character(source, position) if escaped is not None: output.append(escaped[0]) position = escaped[1] continue if source[position] != "{": output.append(source[position]) position += 1 continue closing = matching_brace(source, position) if closing is None: output.append(source[position]) position += 1 continue body = source[position + 1:closing] choices = alternatives(body) if choices is None: inner, inner_used = render(body, depth + 1) output.extend(("{", inner, "}")) used = used or inner_used position = closing + 1 continue if group_number >= max_groups: raise ValueError( "Dynamic prompt contains more than %d alternative groups." % max_groups) selector = hashlib.sha256( ("%d\0%s\0%d\0%s" % (seed, scope, group_number, body)).encode("utf-8") ).digest() group_number += 1 selected = choices[int.from_bytes(selector[:8], "big") % len(choices)] rendered, _nested_used = render(selected, depth + 1) output.append(rendered) used = True position = closing + 1 return "".join(output), used return render(text) def _h3_frame_length(seconds: float) -> int: """Round a duration up to H3's valid 17k+5 frame grid.""" seconds = float(seconds) if not math.isfinite(seconds) or seconds <= 0: raise ValueError("H3 shot duration must be a finite positive number.") # Subtract a tiny tolerance so an exactly frame-aligned decimal does not # jump a frame because of binary floating-point representation. requested = max(5, int(math.ceil(seconds * FPS - 1e-9))) length = requested + (5 - requested % 17) % 17 if length > MAX_H3_FRAMES: raise ValueError( "H3 shot duration %.6fs rounds to %d frames; the largest valid " "17k+5 length is %d frames (%.6fs)." % (seconds, length, MAX_H3_FRAMES, MAX_H3_FRAMES / float(FPS))) return length def _validate_h3_length(length: Any, label: str) -> int: length = int(length) if length < 5 or length > MAX_H3_FRAMES or length % 17 != 5: raise ValueError( "%s must be an H3-valid frame length between 5 and %d " "with length %% 17 == 5; got %d." % (label, MAX_H3_FRAMES, length)) return length def _parse_scene_range(value: Any, total: int, fallback_start: int) -> tuple[int, int]: """Parse one inclusive, contiguous scene selection. Disjoint selections are deliberately rejected: every H3 scene depends on its immediate predecessor, so skipping a scene inside a render selection would either break continuity or silently reuse an invalid checkpoint. """ total = int(total) fallback_start = int(fallback_start) text = str(value or "").strip() if not text: start, end = fallback_start, total else: compact = re.sub(r"\s+", "", text) if "," in compact: raise ValueError( "scene_range supports one contiguous inclusive range only, " "such as '3' or '3:8'. Comma selections are not safe for a " "seamless chain.") match = re.fullmatch(r"(\d+)(?::(\d+))?", compact) if match is None: raise ValueError( "scene_range must be blank, one scene like '3', or one " "inclusive range like '3:8'.") start = int(match.group(1)) end = int(match.group(2) or start) if start < 1 or start > total: raise ValueError("scene_range start must be between 1 and %d." % total) if end < start: raise ValueError("scene_range end must be greater than or equal to start.") if end > total: raise ValueError("scene_range end must be between %d and %d." % (start, total)) return start, end _REFERENCE_TAG_RE = re.compile(r"[A-Za-z][A-Za-z0-9_-]{0,63}") _REFERENCE_ALIAS_RE = re.compile( r"(? list[dict[str, Any]]: return [ { "tag": match.group(1), "timestamp_seconds": ( float(Fraction(match.group(2))) if match.group(2) is not None else None), } for match in _SEMANTIC_ANCHOR_RE.finditer(str(text or "")) ] def _semantic_anchor_mode(value: Any) -> str: mode = str(value or "timestamped_video").strip().lower() if mode not in SEMANTIC_ANCHOR_MODES: raise ValueError( "Semantic anchor mode must be timestamped_video or " "picture_storyboard; got %r." % value) return mode def _tagged_conditioning_backend(value: Any) -> str: backend = str(value or "native_ref2va").strip().lower() if backend not in TAGGED_CONDITIONING_BACKENDS: raise ValueError( "Tagged conditioning backend must be native_ref2va or " "external_refmod; got %r." % value) return backend def _prompt_reference_tags(text: Any) -> set[str]: source = str(text or "") return ( set(_REFERENCE_ALIAS_RE.findall(source)) | {item["tag"] for item in _semantic_anchor_specs(source)} ) def _reference_compliance_mode(value: Any) -> str: # Compatibility with the short-lived BOOLEAN form: True was strict and # False was the warning-only behavior now named soft. if isinstance(value, bool): return "strict" if value else "soft" mode = str(value or "strict").strip().lower() if mode not in REFERENCE_COMPLIANCE_MODES: raise ValueError( "Reference prompt compliance must be strict, soft, or disabled; " "got %r." % value) return mode def _downstream_reference_compliance( dynprompt: Any, unique_id: Any) -> str: """Find the strictest Scheduled Ref2VA policy consuming this schedule. Schedule-builder nodes execute before the wrapper. Looking downstream is therefore the only way an upstream missing source_audio_slice can honor a disabled policy without changing ComfyUI's execution semantics. """ if dynprompt is None or unique_id is None: return "strict" try: node_ids = list(dynprompt.all_node_ids()) except (AttributeError, TypeError): return "strict" queue = [unique_id] visited = {str(unique_id)} modes = [] while queue: parent = queue.pop(0) for node_id in node_ids: if str(node_id) in visited: continue try: node = dynprompt.get_node(node_id) except Exception: continue inputs = node.get("inputs") if isinstance(node, dict) else None if not isinstance(inputs, dict) or not any( isinstance(value, list) and len(value) == 2 and str(value[0]) == str(parent) for value in inputs.values()): continue visited.add(str(node_id)) if node.get("class_type") in ( "MiniMaxH3ScheduledReferenceToVideo", "MiniMaxH3TaggedReferenceToVideo"): try: modes.append(_reference_compliance_mode( inputs.get( "prompt_compliance", inputs.get("reference_policy", "strict")))) except ValueError: modes.append("strict") else: queue.append(node_id) if not modes or "strict" in modes: return "strict" return "soft" if "soft" in modes else "disabled" def _has_downstream_review_gate(dynprompt: Any, unique_id: Any) -> bool: """Return whether this Segment Save directly feeds an H3 Review Gate. Segment Save persists the same video that Review Gate presents. Sending an additional native video-preview payload from the saver duplicates that media in the canvas, has no audio, and can force the saver node to expand to the preview's dimensions. Dynamic prompt IDs are intentionally inspected here so recursive chain iterations behave like the visible graph as well as the initial execution. """ if dynprompt is None or unique_id is None: return False try: node_ids = list(dynprompt.all_node_ids()) except (AttributeError, TypeError): try: original = dynprompt.get_original_prompt() node_ids = list(original) if isinstance(original, dict) else [] except (AttributeError, TypeError): return False for node_id in node_ids: try: node = dynprompt.get_node(node_id) except Exception: continue if (not isinstance(node, dict) or node.get("class_type") != "MiniMaxH3ChainReview"): continue inputs = node.get("inputs") link = inputs.get("segment") if isinstance(inputs, dict) else None if (is_link(link) and str(link[0]) == str(unique_id) and str(link[1]) == "0"): return True return False def _skipped_reference_result( previous: Any, label: str, reason: Any) -> tuple[Any, str, str]: try: schedule = _make_reference_schedule( _reference_schedule_entries(previous)) except (TypeError, ValueError): schedule = _make_reference_schedule([]) message = "%s skipped because compliance is disabled: %s" % ( label, str(reason)) _LOG.warning("H3 scheduled-reference warning: %s", message) return schedule, _reference_fingerprint_output(schedule), message def _skipped_tagged_reference_result( previous: Any, label: str, reason: Any) -> tuple[Any, str, str]: try: references = _make_tagged_references( _tagged_reference_entries(previous)) except (TypeError, ValueError): references = _make_tagged_references([]) message = "%s skipped because reference policy is disabled: %s" % ( label, str(reason)) _LOG.warning("H3 tagged-reference warning: %s", message) return references, _reference_fingerprint_output(references), message def _normalize_reference_tag(value: Any, label: str) -> str: tag = str(value or "").strip() if tag.startswith("@"): tag = tag[1:] if _REFERENCE_TAG_RE.fullmatch(tag) is None: raise ValueError( "%s must be a stable tag such as 'hero_face' or '@hero-face'." % label) return tag def _resolved_media_path(value: Any, label: str) -> str: path = os.path.realpath(os.path.abspath(os.path.expanduser( str(value or "").strip()))) if not str(value or "").strip(): raise ValueError("%s cannot be blank." % label) if not os.path.isfile(path): raise ValueError("%s does not exist or is not a file: %s" % (label, path)) return path def _lazy_motion_media_path( video_path: Any, source_video: Any = None) -> tuple[str, str]: if source_video is not None: get_stream_source = getattr(source_video, "get_stream_source", None) if not callable(get_stream_source): raise ValueError( "Lazy motion source_video must be a native ComfyUI VIDEO " "from a file-backed loader.") try: stream_source = get_stream_source() except Exception as exc: raise ValueError( "Lazy motion source_video could not expose its file source: " "%s" % exc) from exc if not isinstance(stream_source, (str, os.PathLike)): raise ValueError( "Lazy motion source_video is memory-backed. Connect core " "Load Video or provide video_path so scene decoding can " "remain disk-backed.") get_trim = getattr(source_video, "get_active_trim_window", None) if callable(get_trim): try: trim_start, trim_duration = get_trim() except Exception as exc: raise ValueError( "Lazy motion source_video trim could not be inspected: " "%s" % exc) from exc if (abs(float(trim_start)) > 1e-9 or abs(float(trim_duration)) > 1e-9): raise ValueError( "Lazy motion source_video must be untrimmed. Use " "skip_first_frames on Tagged Motion Ref for an exact, " "fingerprinted timeline offset.") return (_resolved_media_path( os.fspath(stream_source), "Lazy motion native VIDEO source"), "native VIDEO loader") return (_resolved_media_path( video_path, "Motion-reference video path"), "direct path") def _stream_start_seconds(stream: Any) -> float: if stream.start_time is None or stream.time_base is None: return 0.0 return float(stream.start_time * stream.time_base) def _stream_duration_seconds( stream: Any, container: Any, fallback_frame_count: int, fallback_fps: float) -> float: """Prefer media timestamps over a container's advisory frame count.""" if stream.duration is not None and stream.time_base is not None: duration = float(stream.duration * stream.time_base) if math.isfinite(duration) and duration > 0.0: return duration if container.duration is not None: duration = float(container.duration) / float(av.time_base) if math.isfinite(duration) and duration > 0.0: return duration if int(fallback_frame_count) > 0 and float(fallback_fps) > 0.0: return int(fallback_frame_count) / float(fallback_fps) return 0.0 def _probe_lazy_motion_path( path: str, use_embedded_audio: bool) -> dict[str, Any]: if av is None: raise RuntimeError("Lazy H3 motion references require PyAV.") resolved = _resolved_media_path(path, "Motion-reference video path") try: with av.open(resolved, mode="r") as container: videos = list(container.streams.video) if len(videos) != 1: raise ValueError( "Lazy motion reference %s contains %d video streams; " "exactly one is required." % (resolved, len(videos))) video = videos[0] rate = video.average_rate or video.base_rate or video.guessed_rate if rate is None or not math.isfinite(float(rate)): raise ValueError( "Lazy motion reference has no usable frame rate.") source_rate = Fraction(rate) source_fps = float(source_rate) if source_fps <= 0.0: raise ValueError( "Lazy motion reference has an invalid frame rate %.6g." % source_fps) width = int(video.codec_context.width or 0) height = int(video.codec_context.height or 0) if width < 1 or height < 1: raise ValueError( "Lazy motion reference has invalid video dimensions.") source_frame_count = int(video.frames or 0) duration_seconds = _stream_duration_seconds( video, container, source_frame_count, source_fps) frame_count = (int(math.ceil( duration_seconds * FPS - 1e-9)) if duration_seconds > 0.0 else 0) audios = list(container.streams.audio) audio_info = None if bool(use_embedded_audio): if len(audios) != 1: raise ValueError( "Lazy motion reference requested embedded audio, but " "%s contains %d audio streams; exactly one is " "required." % (resolved, len(audios))) audio = audios[0] sample_rate = int( audio.codec_context.sample_rate or audio.rate or 0) channels = len(audio.codec_context.layout.channels) if sample_rate < 1 or channels not in (1, 2): raise ValueError( "Lazy motion-reference audio must be mono or stereo " "with a valid sample rate.") audio_info = { "sample_rate": sample_rate, "channels": channels, "start_seconds": _stream_start_seconds(audio), } return { "version": LAZY_MOTION_SOURCE_VERSION, "kind": "lazy_motion_path", "path": resolved, "fps": FPS, "source_fps": source_fps, "source_rate_numerator": int(source_rate.numerator), "source_rate_denominator": int(source_rate.denominator), "width": width, "height": height, "frame_count": frame_count, "source_frame_count": source_frame_count, "duration_seconds": duration_seconds, "video_start_seconds": _stream_start_seconds(video), "audio": audio_info, } except ValueError: raise except Exception as exc: raise ValueError( "Could not inspect lazy motion-reference video %s: %s" % (resolved, exc)) from exc def _probe_audio_path( path: str, label: str = "Source Timeline audio path", required: bool = True) -> dict[str, Any] | None: """Inspect one path-backed audio stream without decoding its samples.""" if av is None: raise RuntimeError("H3 Source Timeline audio paths require PyAV.") resolved = _resolved_media_path(path, label) try: with av.open(resolved, mode="r") as container: audios = list(container.streams.audio) if not audios and not bool(required): return None if len(audios) != 1: raise ValueError( "%s contains %d audio streams; exactly one is required." % (label, len(audios))) audio = audios[0] sample_rate = int( audio.codec_context.sample_rate or audio.rate or 0) channels = len(audio.codec_context.layout.channels) if sample_rate < 1 or channels not in (1, 2): raise ValueError( "%s must be mono or stereo with a valid sample rate." % label) duration_seconds = _stream_duration_seconds( audio, container, 0, float(sample_rate)) return { "path": resolved, "stream_index": int(audio.index), "sample_rate": sample_rate, "channels": channels, "start_seconds": _stream_start_seconds(audio), "duration_seconds": duration_seconds, } except ValueError: raise except Exception as exc: raise ValueError( "Could not inspect %s %s: %s" % (label, resolved, exc)) from exc def _lazy_motion_descriptor_with_skip( descriptor: dict[str, Any], skip_first_frames: Any) -> dict[str, Any]: prepared = dict(descriptor) skip_first_frames = int(skip_first_frames) if skip_first_frames < 0: raise ValueError( "Lazy motion skip_first_frames cannot be negative.") source_frames = int(prepared.get("source_frame_count", 0)) if source_frames > 0 and skip_first_frames >= source_frames: raise ValueError( "Lazy motion skip_first_frames %d consumes all %d source " "frames." % (skip_first_frames, source_frames)) source_fps = float(prepared["source_fps"]) duration_seconds = float(prepared.get("duration_seconds", 0.0)) skip_seconds = skip_first_frames / source_fps if duration_seconds > 0.0 and skip_seconds >= duration_seconds: raise ValueError( "Lazy motion skip_first_frames %d starts at or beyond the " "%.3fs source duration." % (skip_first_frames, duration_seconds)) prepared["skip_first_frames"] = skip_first_frames prepared["skip_seconds"] = skip_seconds if duration_seconds > 0.0: prepared["frame_count"] = int(math.ceil( (duration_seconds - skip_seconds) * FPS - 1e-9)) return prepared def _is_lazy_motion_descriptor(value: Any) -> bool: if not isinstance(value, dict): return False try: version = int(value.get("version", -1)) except (TypeError, ValueError): return False return (version == LAZY_MOTION_SOURCE_VERSION and value.get("kind") == "lazy_motion_path" and isinstance(value.get("path"), str)) PROJECT_ASSET_DESCRIPTOR_VERSION = 1 def _project_asset_descriptor(entry: dict[str, Any], path: str) -> dict[str, Any]: """Small runtime handle; media bytes never enter workflow serialization.""" return { "version": PROJECT_ASSET_DESCRIPTOR_VERSION, "kind": "project_asset_path", "path": os.path.realpath(path), "asset_id": str(entry.get("id") or ""), "media_kind": str(entry.get("kind") or ""), "content_hash": str(entry.get("sha256") or ""), } def _is_project_asset_descriptor(value: Any, media_kind: str = "") -> bool: if not isinstance(value, dict): return False try: version = int(value.get("version", -1)) except (TypeError, ValueError): return False return (version == PROJECT_ASSET_DESCRIPTOR_VERSION and value.get("kind") == "project_asset_path" and isinstance(value.get("path"), str) and (not media_kind or value.get("media_kind") == media_kind)) def _project_asset_media_path(value: Any, media_kind: str) -> str: if not _is_project_asset_descriptor(value, media_kind): raise ValueError("Invalid H3 project %s asset descriptor." % media_kind) return _resolved_media_path( value["path"], "H3 project %s asset" % media_kind) def _project_asset_image(value: Any) -> Any: if not _is_project_asset_descriptor(value, "image"): return value if Image is None or ImageOps is None or np is None or torch is None: raise RuntimeError( "H3 project picture references require Pillow, NumPy, and torch.") path = _project_asset_media_path(value, "image") with Image.open(path) as source: image = ImageOps.exif_transpose(source).convert("RGB") array = np.asarray(image, dtype=np.float32) / 255.0 return torch.from_numpy(array.copy()).unsqueeze(0) def _project_asset_audio(value: Any) -> Any: if not _is_project_asset_descriptor(value, "audio"): return value path = _project_asset_media_path(value, "audio") timeline = _make_source_timeline( audio_path=path, embedded_audio="ignore", source_route="H3 project asset") return _source_timeline_source_audio(timeline) def _project_asset_video_descriptor( entry: dict[str, Any], path: str, *, embedded_audio: bool, short_edge: str) -> dict[str, Any]: metadata = entry.get("metadata") metadata = metadata if isinstance(metadata, dict) else {} source_fps = float(metadata.get("fps") or FPS) numerator = int(metadata.get("source_rate_numerator") or round(source_fps)) denominator = int(metadata.get("source_rate_denominator") or 1) duration = float(metadata.get("duration_seconds") or 0.0) descriptor = { "version": LAZY_MOTION_SOURCE_VERSION, "kind": "lazy_motion_path", "path": os.path.realpath(path), "fps": FPS, "source_fps": source_fps, "source_rate_numerator": numerator, "source_rate_denominator": denominator, "width": int(metadata.get("width") or 0), "height": int(metadata.get("height") or 0), "frame_count": int(metadata.get("frame_count") or ( math.ceil(duration * FPS - 1e-9) if duration > 0 else 0)), "source_frame_count": int(metadata.get("frame_count") or 0), "duration_seconds": duration, "video_start_seconds": float(metadata.get("video_start_seconds") or 0), "skip_first_frames": 0, "skip_seconds": 0.0, "motion_short_edge": str(short_edge), "file_sha256": str(entry.get("sha256") or ""), "source_route": "H3 project asset", "audio": None, } if (descriptor["width"] < 1 or descriptor["height"] < 1 or descriptor["source_fps"] <= 0): descriptor = _probe_lazy_motion_path(path, bool(embedded_audio)) descriptor["motion_short_edge"] = str(short_edge) descriptor["file_sha256"] = str(entry.get("sha256") or "") descriptor["source_route"] = "H3 project asset" return descriptor if embedded_audio: sample_rate = int(metadata.get("audio_sample_rate") or 0) channels = int(metadata.get("audio_channels") or 0) if sample_rate < 1 or channels not in (1, 2): descriptor = _probe_lazy_motion_path(path, True) descriptor["motion_short_edge"] = str(short_edge) descriptor["file_sha256"] = str(entry.get("sha256") or "") descriptor["source_route"] = "H3 project asset" return descriptor descriptor["audio"] = { "sample_rate": sample_rate, "channels": channels, "start_seconds": float(metadata.get("audio_start_seconds") or 0), } return descriptor def _parse_reference_selector( value: Any, total: int | None = None) -> tuple[tuple[int, int], ...]: """Parse a disjoint, one-based scene selector and merge overlaps.""" text = re.sub(r"\s+", "", str(value or "")).lower() if text in ("", "*", "all"): return () ranges = [] for token in text.split(","): match = re.fullmatch(r"(\d+)(?::(\d+))?", token) if match is None: raise ValueError( "Reference scenes must be blank/all, one scene like '3', " "or comma-separated inclusive ranges like '1,3,5:8'.") start = int(match.group(1)) end = int(match.group(2) or start) if start < 1: raise ValueError("Reference scene numbers start at 1.") if end < start: raise ValueError( "Reference range %s ends before it starts." % token) if total is not None and end > int(total): raise ValueError( "Reference range %s exceeds this plan's %d scenes." % (token, int(total))) ranges.append((start, end)) merged = [] for start, end in sorted(ranges): if merged and start <= merged[-1][1] + 1: merged[-1] = (merged[-1][0], max(merged[-1][1], end)) else: merged.append((start, end)) return tuple(merged) def _reference_selector_text(ranges: Any) -> str: if not ranges: return "all" return ",".join( str(start) if int(start) == int(end) else "%d:%d" % (start, end) for start, end in ranges) def _reference_is_active(entry: dict[str, Any], scene: int) -> bool: ranges = entry.get("ranges") or () return not ranges or any( int(start) <= int(scene) <= int(end) for start, end in ranges) def _reference_entry_contract(entry: dict[str, Any]) -> dict[str, Any]: keys = ( "kind", "tag", "scenes", "content_hash", "audio_tag", "audio_hash", "paired_audio_policy", "semantic_role", "motion_target", "motion_description", "motion_short_edge", "semantic_anchor_size", "semantic_anchor_mode", ) contract = {key: entry[key] for key in keys if key in entry} # Keep old schedules bit-identical. Timeline metadata enters the resume # contract only when a reference opts into non-default playback. default_timeline = ( "standalone" if entry.get("kind") == "audio" else "restart_each_scene") timeline_mode = str(entry.get("timeline_mode") or default_timeline) if timeline_mode != default_timeline: contract["timeline_mode"] = timeline_mode if bool(entry.get("align_audio_reference")): contract["align_audio_reference"] = True if str(entry.get("activation") or "schedule") != "schedule": contract["activation"] = str(entry["activation"]) return contract def _reference_schedule_entries(value: Any) -> list[dict[str, Any]]: if value is None: return [] if (not isinstance(value, dict) or int(value.get("version", -1)) != REFERENCE_SCHEDULE_VERSION or not isinstance(value.get("entries"), list)): raise ValueError( "Scheduled references must come from this pack's Picture, Video, " "or Audio Schedule nodes.") return list(value["entries"]) def _reference_entry_tags(entry: dict[str, Any]) -> tuple[str, ...]: tags = [str(entry["tag"])] if entry.get("audio_tag"): tags.append(str(entry["audio_tag"])) return tuple(tags) def _make_reference_schedule( entries: list[dict[str, Any]]) -> dict[str, Any]: contracts = [_reference_entry_contract(entry) for entry in entries] return { "version": REFERENCE_SCHEDULE_VERSION, "entries": entries, "fingerprint": _fingerprint({ "version": REFERENCE_SCHEDULE_VERSION, "entries": contracts, }), } def _reference_fingerprint_lineage(value: Any) -> dict[str, Any]: """Return JSON-safe history and contracts for one reference registry.""" entries = _reference_schedule_entries(value) contracts = [_reference_entry_contract(entry) for entry in entries] registry_mode = ( "tagged" if isinstance(value, dict) and value.get("activation") == "prompt" else "ordered") prefixes = [{ "fingerprint": _reference_contracts_fingerprint( [], registry_mode=registry_mode), "entry": None, }] for index, entry in enumerate(contracts, start=1): prefixes.append({ "fingerprint": _reference_contracts_fingerprint( contracts[:index], registry_mode=registry_mode), "entry": entry, }) current = str(value.get("fingerprint") or "") if not current or prefixes[-1]["fingerprint"] != current: raise ValueError("H3 reference registry fingerprint is inconsistent.") return { "version": REFERENCE_FINGERPRINT_LINEAGE_VERSION, "current": current, "prefixes": prefixes, "registry_mode": registry_mode, # Full contracts let resume recognize a previously saved ordered # registry when newly added, inactive nodes were inserted anywhere in # the ComfyUI chain rather than only attached at its output end. "entries": contracts, } def _reference_fingerprint_output( value: Any, *, wrapper_key: str = "", wrapper_contract: dict[str, Any] | None = None) -> str: """Encode the fingerprint plus safe append ancestry in a STRING socket.""" lineage = _reference_fingerprint_lineage(value) if wrapper_key: extras = dict(wrapper_contract or {}) prefixes = [] for item in lineage["prefixes"]: fingerprint = _fingerprint({ wrapper_key: item["fingerprint"], **extras, }) prefixes.append({ "fingerprint": fingerprint, "entry": item["entry"], }) lineage = { "version": REFERENCE_FINGERPRINT_LINEAGE_VERSION, "current": prefixes[-1]["fingerprint"], "prefixes": prefixes, "registry_mode": lineage["registry_mode"], "entries": lineage["entries"], "wrapper": { "key": wrapper_key, "contract": extras, }, } return _canonical_json({ "h3_reference_fingerprint_lineage": lineage, }) def _reference_contracts_fingerprint( entries: list[dict[str, Any]], wrapper: dict[str, Any] | None = None, registry_mode: str = "ordered") -> str: contracts = list(entries) if str(registry_mode) == "tagged": contracts = sorted(contracts, key=lambda entry: ( str(entry.get("kind") or ""), str(entry.get("tag") or ""), str(entry.get("audio_tag") or ""), _canonical_json(entry), )) fingerprint = _fingerprint({ "version": REFERENCE_SCHEDULE_VERSION, "entries": contracts, }) if isinstance(wrapper, dict): fingerprint = _fingerprint({ str(wrapper["key"]): fingerprint, **dict(wrapper.get("contract") or {}), }) return fingerprint def _generation_fingerprint_value( value: Any) -> tuple[str, dict[str, Any] | None]: """Decode a reference-lineage token or preserve a manual fingerprint.""" text = str(value or "").strip() if not text.startswith("{"): return text, None try: document = json.loads(text) except json.JSONDecodeError: return text, None lineage = (document.get("h3_reference_fingerprint_lineage") if isinstance(document, dict) else None) if not isinstance(lineage, dict): return text, None if int(lineage.get("version", -1)) != ( REFERENCE_FINGERPRINT_LINEAGE_VERSION): raise ValueError("Unsupported H3 reference fingerprint lineage version.") current = str(lineage.get("current") or "") prefixes = lineage.get("prefixes") if (not re.fullmatch(r"[0-9a-f]{64}", current) or not isinstance(prefixes, list) or not prefixes or not isinstance(prefixes[-1], dict) or str(prefixes[-1].get("fingerprint") or "") != current): raise ValueError("H3 reference fingerprint lineage is malformed.") normalized = [] for index, item in enumerate(prefixes): if not isinstance(item, dict): raise ValueError("H3 reference fingerprint lineage is malformed.") fingerprint = str(item.get("fingerprint") or "") entry = item.get("entry") if not re.fullmatch(r"[0-9a-f]{64}", fingerprint): raise ValueError("H3 reference fingerprint lineage is malformed.") if index == 0: if entry is not None: raise ValueError( "H3 reference fingerprint lineage base must be empty.") elif not isinstance(entry, dict): raise ValueError("H3 reference fingerprint lineage is malformed.") normalized.append({"fingerprint": fingerprint, "entry": entry}) entries = lineage.get("entries") if entries is not None and ( not isinstance(entries, list) or any(not isinstance(entry, dict) for entry in entries)): raise ValueError("H3 reference fingerprint lineage is malformed.") normalized_lineage = { "version": REFERENCE_FINGERPRINT_LINEAGE_VERSION, "current": current, "prefixes": normalized, } registry_mode = str(lineage.get("registry_mode") or "ordered") if registry_mode not in ("ordered", "tagged"): raise ValueError("H3 reference fingerprint lineage is malformed.") normalized_lineage["registry_mode"] = registry_mode if entries is not None: normalized_lineage["entries"] = _json_document(entries) wrapper = lineage.get("wrapper") if wrapper is not None: if (not isinstance(wrapper, dict) or not isinstance(wrapper.get("key"), str) or not wrapper["key"] or not isinstance(wrapper.get("contract"), dict)): raise ValueError("H3 reference fingerprint lineage is malformed.") normalized_lineage["wrapper"] = _json_document(wrapper) if entries is not None: if (_reference_contracts_fingerprint( normalized_lineage["entries"], normalized_lineage.get("wrapper"), registry_mode) != current): raise ValueError( "H3 reference fingerprint lineage does not match its entries.") for index, item in enumerate(normalized): if (_reference_contracts_fingerprint( normalized_lineage["entries"][:index], normalized_lineage.get("wrapper"), registry_mode) != item["fingerprint"]): raise ValueError( "H3 reference fingerprint lineage has invalid history.") return current, normalized_lineage def _tagged_reference_entries(value: Any) -> list[dict[str, Any]]: entries = _reference_schedule_entries(value) if not isinstance(value, dict) or value.get("activation") != "prompt": raise ValueError( "Tagged references must come from this pack's Tagged Picture, " "Video, or Audio Ref nodes.") if any(entry.get("activation") != "prompt" for entry in entries): raise ValueError( "A tagged-reference chain cannot contain legacy scheduled " "entries. Use one node family or the other.") return entries def _make_tagged_references( entries: list[dict[str, Any]]) -> dict[str, Any]: result = _make_reference_schedule(entries) result["activation"] = "prompt" result["fingerprint"] = _reference_contracts_fingerprint( [_reference_entry_contract(entry) for entry in entries], registry_mode="tagged") return result def _semantic_anchor_draft_entries(value: Any) -> list[dict[str, Any]]: if (not isinstance(value, dict) or int(value.get("version", -1)) != SEMANTIC_ANCHOR_REGISTRY_VERSION or value.get("kind") not in ("draft", "bundle") or not isinstance(value.get("entries"), list)): raise ValueError( "Semantic anchors must come from this pack's Semantic Picture " "Anchor nodes.") entries = list(value["entries"]) for entry in entries: if (not isinstance(entry, dict) or entry.get("kind") != "semantic_anchor" or not isinstance(entry.get("tag"), str) or not entry.get("tag") or "value" not in entry or not isinstance(entry.get("content_hash"), str)): raise ValueError("Semantic anchor registry is malformed.") return entries def _make_semantic_anchor_draft( entries: list[dict[str, Any]]) -> dict[str, Any]: contracts = [_reference_entry_contract(entry) for entry in entries] return { "version": SEMANTIC_ANCHOR_REGISTRY_VERSION, "kind": "draft", "entries": entries, "fingerprint": _reference_contracts_fingerprint( contracts, registry_mode="tagged"), } def _semantic_anchor_bundle(value: Any) -> dict[str, Any]: entries = _semantic_anchor_draft_entries(value) if value.get("kind") != "bundle": raise ValueError( "Connect Semantic Picture Anchors through Semantic Anchor Bundle " "before using them in Ref2VA, Plan Studio, or Run Manager.") size = str(value.get("semantic_anchor_size") or "512") mode = str(value.get("semantic_anchor_mode") or "timestamped_video").strip().lower() if size not in SEMANTIC_ANCHOR_BUNDLE_SIZES: raise ValueError( "Semantic anchor size must be one of %s." % (SEMANTIC_ANCHOR_BUNDLE_SIZES,)) if mode not in SEMANTIC_ANCHOR_BUNDLE_MODES: raise ValueError( "Semantic anchor mode must be one of %s." % (SEMANTIC_ANCHOR_BUNDLE_MODES,)) expected = _make_semantic_anchor_bundle(entries, size, mode) if str(value.get("fingerprint") or "") != expected["fingerprint"]: raise ValueError("Semantic anchor bundle fingerprint is inconsistent.") return value def _make_semantic_anchor_bundle( entries: list[dict[str, Any]], size: Any, mode: Any) -> dict[str, Any]: size = str(size or "512") mode = str(mode or "timestamped_video").strip().lower() if size not in SEMANTIC_ANCHOR_BUNDLE_SIZES: raise ValueError( "Semantic anchor size must be one of %s." % (SEMANTIC_ANCHOR_BUNDLE_SIZES,)) if mode not in SEMANTIC_ANCHOR_BUNDLE_MODES: raise ValueError( "Semantic anchor mode must be one of %s." % (SEMANTIC_ANCHOR_BUNDLE_MODES,)) bundled_entries = [] for source in entries: entry = dict(source) entry["semantic_anchor_size"] = size entry["semantic_anchor_mode"] = mode bundled_entries.append(entry) contracts = [_reference_entry_contract(entry) for entry in bundled_entries] return { "version": SEMANTIC_ANCHOR_REGISTRY_VERSION, "kind": "bundle", "entries": bundled_entries, "semantic_anchor_size": size, "semantic_anchor_mode": mode, "fingerprint": _reference_contracts_fingerprint( contracts, registry_mode="tagged"), } def _reference_semantic_anchor_bundle( references: Any) -> dict[str, Any] | None: """Resolve Qwen anchors embedded in the tagged reference carrier.""" carried = ( references.get("semantic_anchors") if isinstance(references, dict) else None) return _semantic_anchor_bundle(carried) if carried is not None else None def _combined_reference_registry( references: Any, semantic_anchors: Any = None) -> dict[str, Any]: """Build one lineage token while keeping native and Qwen media separate.""" entries = _tagged_reference_entries(references) bundle = ( _semantic_anchor_bundle(semantic_anchors) if semantic_anchors is not None else _reference_semantic_anchor_bundle(references)) if bundle is not None: entries = entries + _semantic_anchor_draft_entries(bundle) return _make_tagged_references(entries) def _plan_studio_generation_fingerprint( generation_fingerprint: Any, tagged_references: Any) -> str: """Compose Studio's external contract with its connected references.""" external = str(generation_fingerprint or "").strip() if tagged_references is None: return external combined = _combined_reference_registry(tagged_references) if not combined["entries"]: # Merely connecting an empty reference line must not change an older # standalone Plan's compatibility contract. return external reference_token = _reference_fingerprint_output(combined) external_current, external_lineage = _generation_fingerprint_value( external) if external_lineage is not None and ( external_current == combined["fingerprint"]): # Existing workflows may still wire the reference STRING output into # the legacy field. Avoid wrapping the same lineage a second time. return reference_token if not external: return reference_token return _reference_fingerprint_output( combined, wrapper_key="tagged_reference_fingerprint", wrapper_contract={ "external_generation_fingerprint": external, }) def _append_scheduled_reference( previous: Any, *, kind: str, tag: Any, scenes: Any, value: Any, content_hash: str, audio: Any = None, audio_tag: Any = "", audio_hash: str = "", compliance_mode: str = "strict", timeline_mode: Any = None, align_audio_reference: Any = False, paired_audio_policy: Any = None) -> dict[str, Any]: compliance = _reference_compliance_mode(compliance_mode) entries = _reference_schedule_entries(previous) try: normalized_tag = _normalize_reference_tag(tag, "Reference tag") except ValueError as exc: if compliance != "disabled": raise normalized_tag = "reference_%s" % str(content_hash)[:12] _LOG.warning( "H3 scheduled-reference warning: %s Using internal tag @%s.", exc, normalized_tag) try: ranges = _parse_reference_selector(scenes) except ValueError as exc: if compliance != "disabled": raise ranges = () _LOG.warning( "H3 scheduled-reference warning: %s Treating the reference as " "active in every scene.", exc) entry = { "kind": str(kind), "tag": normalized_tag, "scenes": _reference_selector_text(ranges), "ranges": ranges, "value": value, "content_hash": str(content_hash), } if kind == "video": normalized_timeline = str( timeline_mode or "restart_each_scene").strip().lower() if normalized_timeline not in REFERENCE_VIDEO_TIMELINE_MODES: raise ValueError( "Scheduled video timeline_mode must be one of %s." % (REFERENCE_VIDEO_TIMELINE_MODES,)) entry["timeline_mode"] = normalized_timeline elif kind == "audio": normalized_timeline = str( timeline_mode or "standalone").strip().lower() if normalized_timeline not in REFERENCE_AUDIO_TIMELINE_MODES: raise ValueError( "Scheduled audio timeline_mode must be one of %s." % (REFERENCE_AUDIO_TIMELINE_MODES,)) entry["timeline_mode"] = normalized_timeline entry["align_audio_reference"] = bool(align_audio_reference) if audio is not None: try: normalized_audio_tag = _normalize_reference_tag( audio_tag or (normalized_tag + "_audio"), "Paired video-audio tag") except ValueError as exc: if compliance != "disabled": raise normalized_audio_tag = "%s_audio" % normalized_tag _LOG.warning( "H3 scheduled-reference warning: %s Using internal tag @%s.", exc, normalized_audio_tag) entry.update({ "audio": audio, "audio_tag": normalized_audio_tag, "audio_hash": str(audio_hash), }) if kind == "video" and paired_audio_policy is not None: paired_policy = _paired_audio_policy(paired_audio_policy) if paired_policy == "embedded" and audio is None: raise ValueError( "paired_audio_policy=embedded requires paired audio on the " "video reference.") if paired_policy == "off" and audio is not None: raise ValueError( "paired_audio_policy=off cannot carry paired audio.") entry["paired_audio_policy"] = paired_policy existing_tags = { alias for existing in entries for alias in _reference_entry_tags(existing) } new_tags = _reference_entry_tags(entry) if compliance != "disabled" and len(set(new_tags)) != len(new_tags): raise ValueError( "A scheduled video's @tag and paired @audio_tag must be " "different.") duplicates = existing_tags.intersection(new_tags) if compliance != "disabled" and duplicates: duplicate = sorted(duplicates)[0] raise ValueError( "Scheduled reference tag @%s is already in this chain." % duplicate) return _make_reference_schedule(entries + [entry]) def _append_tagged_reference( previous: Any, *, kind: str, tag: Any, value: Any, content_hash: str, audio: Any = None, audio_tag: Any = "", audio_hash: str = "", compliance_mode: str = "strict", timeline_mode: Any = None, align_audio_reference: Any = False, paired_audio_policy: Any = None) -> dict[str, Any]: if previous is None: seed = None else: seed = _make_reference_schedule(_tagged_reference_entries(previous)) collection = _append_scheduled_reference( seed, kind=kind, tag=tag, scenes="all", value=value, content_hash=content_hash, audio=audio, audio_tag=audio_tag, audio_hash=audio_hash, compliance_mode=compliance_mode, timeline_mode=timeline_mode, align_audio_reference=align_audio_reference, paired_audio_policy=paired_audio_policy) entries = collection["entries"] entries[-1]["activation"] = "prompt" return _make_tagged_references(entries) def _lazy_motion_target_size( descriptor: dict[str, Any]) -> tuple[int, int]: source_w = int(descriptor["width"]) source_h = int(descriptor["height"]) short_edge = str(descriptor.get("motion_short_edge") or "384") if short_edge == "source" or min(source_w, source_h) <= int(short_edge): return source_w, source_h scale = int(short_edge) / float(min(source_w, source_h)) target_h = max(32, round(source_h * scale / 32) * 32) target_w = max(32, round(source_w * scale / 32) * 32) return target_w, target_h def _decode_lazy_motion_video( descriptor: dict[str, Any], source_start: int, source_end: int) -> Any: if av is None or np is None or torch is None: raise RuntimeError( "Lazy H3 motion video decoding requires PyAV, NumPy, and torch.") path = _resolved_media_path( descriptor.get("path"), "Lazy motion-reference video path") source_start = int(source_start) source_end = int(source_end) requested = source_end - source_start if source_start < 0 or requested < 5: raise ValueError( "Lazy motion-reference 24 fps target window must contain at " "least five frames; got %d:%d." % (source_start, source_end)) source_fps = float(descriptor.get("source_fps", FPS)) if not math.isfinite(source_fps) or source_fps <= 0.0: raise ValueError( "Lazy motion reference has invalid source_fps %.6g." % source_fps) skip_first_frames = int(descriptor.get("skip_first_frames", 0)) if skip_first_frames < 0: raise ValueError( "Lazy motion reference skip_first_frames cannot be negative.") skip_seconds = skip_first_frames / source_fps target_w, target_h = _lazy_motion_target_size(descriptor) buffer = np.empty( (requested, target_h, target_w, 3), dtype=np.uint8) video_start = float(descriptor.get("video_start_seconds", 0.0)) def decode_window(allow_seek: bool) -> int | None: decoded = 0 with av.open(path, mode="r") as container: videos = list(container.streams.video) if len(videos) != 1: raise ValueError( "Lazy motion reference changed and now contains %d " "video streams; exactly one is required." % len(videos)) video = videos[0] actual_rate = ( video.average_rate or video.base_rate or video.guessed_rate) if (actual_rate is None or abs(float(actual_rate) - source_fps) > 1e-3): raise ValueError( "Lazy motion reference changed frame rate after it was " "registered (expected %.6g fps, got %s)." % (source_fps, actual_rate)) target_start_seconds = ( skip_seconds + source_start / float(FPS)) seeking = bool( allow_seek and video.time_base is not None and target_start_seconds > 2.0 / source_fps) if seeking: seek_seconds = video_start + max( 0.0, target_start_seconds - 2.0 / source_fps) seek_offset = int(math.floor( seek_seconds / float(video.time_base))) container.seek( seek_offset, stream=video, backward=True, any_frame=False) previous_frame = None previous_time = None previous_array = None target_index = source_start target_tolerance = 1e-7 def frame_array(frame): if (int(frame.width) != target_w or int(frame.height) != target_h): converted = frame.reformat( width=target_w, height=target_h, format="rgb24") return converted.to_ndarray() return frame.to_ndarray(format="rgb24") def emit_previous(): nonlocal decoded, previous_array if previous_frame is None: raise ValueError( "Lazy motion reference starts after its first 24 fps " "target timestamp.") if previous_array is None: previous_array = frame_array(previous_frame) buffer[decoded] = previous_array decoded += 1 for local_index, frame in enumerate(container.decode(video)): if frame.pts is not None and frame.time_base is not None: timestamp = float(frame.pts * frame.time_base) timestamp_relative = timestamp - video_start source_position = timestamp_relative * source_fps frame_index = int(round(source_position)) pts_tolerance = max( 0.05, abs(float(frame.time_base)) * source_fps * 0.6) if abs(source_position - frame_index) > pts_tolerance: raise ValueError( "Lazy motion reference is not constant %.6g fps " "near decoded timestamp %.6fs (source position " "%.4f)." % (source_fps, timestamp, source_position)) elif seeking: # Exact global indexing is unavailable after a keyframe # seek. Reopen and decode from frame zero as a safe # fallback; RAM remains bounded even for this rare case. return None else: frame_index = local_index # Once CFR has been verified, use its rational global frame # index rather than coarse container PTS for target selection. relative_time = frame_index / source_fps target_time = ( skip_seconds + target_index / float(FPS)) while (target_index < source_end and target_time < relative_time - target_tolerance): if (previous_time is None or target_time < previous_time - target_tolerance): raise ValueError( "Lazy motion reference cannot cover 24 fps target " "frame %d after skipping %d source frames." % (target_index, skip_first_frames)) emit_previous() target_index += 1 target_time = ( skip_seconds + target_index / float(FPS)) if target_index >= source_end: break previous_frame = frame previous_time = relative_time previous_array = None if target_index < source_end and previous_frame is not None: display_end = previous_time + 1.0 / source_fps target_time = ( skip_seconds + target_index / float(FPS)) while (target_index < source_end and target_time < display_end - target_tolerance): emit_previous() target_index += 1 target_time = ( skip_seconds + target_index / float(FPS)) return decoded decoded = decode_window(allow_seek=True) if decoded is None: decoded = decode_window(allow_seek=False) if decoded != requested: raise ValueError( "Lazy motion reference %s yielded only %d of %d requested " "frames for 24 fps target window %d:%d after skipping %d source " "frames." % (path, decoded, requested, source_start, source_end, skip_first_frames)) return torch.from_numpy(buffer).to(dtype=torch.float32).div_(255.0) def _resampled_audio_frames(resampler: Any, frame: Any): values = resampler.resample(frame) if values is None: return () if isinstance(values, (list, tuple)): return values return (values,) def _decode_lazy_motion_audio( descriptor: dict[str, Any], source_start: int, source_end: int) -> dict[str, Any] | None: audio_info = descriptor.get("audio") if audio_info is None: return None if av is None or np is None or torch is None: raise RuntimeError( "Lazy H3 motion audio decoding requires PyAV, NumPy, and torch.") path = _resolved_media_path( descriptor.get("path"), "Lazy motion-reference video path") sample_rate = int(audio_info["sample_rate"]) channels = int(audio_info["channels"]) layout = "mono" if channels == 1 else "stereo" video_start = float(descriptor.get("video_start_seconds", 0.0)) audio_start = float(audio_info.get("start_seconds", 0.0)) source_fps = float(descriptor.get("source_fps", FPS)) skip_first_frames = int(descriptor.get("skip_first_frames", 0)) skip_seconds = skip_first_frames / source_fps start_seconds = ( video_start + skip_seconds + int(source_start) / float(FPS)) end_seconds = ( video_start + skip_seconds + int(source_end) / float(FPS)) sample_start = int(round((start_seconds - audio_start) * sample_rate)) sample_end = int(round((end_seconds - audio_start) * sample_rate)) if sample_start < 0 or sample_end <= sample_start: raise ValueError( "Lazy motion-reference audio cannot cover video window %d:%d " "because its stream starts at %.6fs." % (int(source_start), int(source_end), audio_start)) requested = sample_end - sample_start target = np.zeros((channels, requested), dtype=np.float32) coverage = np.zeros((requested,), dtype=np.bool_) cursor = 0 cursor_initialized = False with av.open(path, mode="r") as container: audios = list(container.streams.audio) if len(audios) != 1: raise ValueError( "Lazy motion reference changed and now contains %d audio " "streams; exactly one is required." % len(audios)) audio_stream = audios[0] seek_preroll = max(0.25, 2.0 / source_fps) audio_seek = bool( audio_stream.time_base is not None and start_seconds - audio_start > seek_preroll) if audio_seek: seek_seconds = max( audio_start, start_seconds - seek_preroll) seek_offset = int(math.floor( seek_seconds / float(audio_stream.time_base))) container.seek( seek_offset, stream=audio_stream, backward=True, any_frame=False) resampler = av.AudioResampler( format="fltp", layout=layout, rate=sample_rate) def consume(output_frame): nonlocal cursor, cursor_initialized array = output_frame.to_ndarray() if array.ndim == 1: array = array.reshape(1, -1) if tuple(array.shape[:1]) != (channels,): raise ValueError( "Lazy motion-reference audio decoded to unexpected shape " "%s." % (tuple(array.shape),)) count = int(array.shape[-1]) if not cursor_initialized: if (output_frame.pts is not None and output_frame.time_base is not None): frame_seconds = float( output_frame.pts * output_frame.time_base) cursor = int(round( (frame_seconds - audio_start) * sample_rate)) elif audio_seek: raise ValueError( "Lazy motion-reference audio has no timestamps after " "seeking; exact skip alignment is unavailable.") cursor_initialized = True copy_start = max(cursor, sample_start) copy_end = min(cursor + count, sample_end) if copy_end > copy_start: source_offset = copy_start - cursor target_offset = copy_start - sample_start span = copy_end - copy_start target[:, target_offset:target_offset + span] = ( array[:, source_offset:source_offset + span]) coverage[target_offset:target_offset + span] = True cursor += count stop = False for frame in container.decode(audio_stream): for output_frame in _resampled_audio_frames(resampler, frame): consume(output_frame) if cursor >= sample_end: stop = True break if stop: break if not stop: for output_frame in _resampled_audio_frames(resampler, None): consume(output_frame) if cursor >= sample_end: break copied = int(np.count_nonzero(coverage)) if copied != requested: first_missing = int(np.flatnonzero(~coverage)[0]) internal_gap = bool(np.any(coverage[first_missing:])) tail_missing = requested - first_missing maximum_rounding_pad = int(math.ceil(sample_rate / float(FPS))) if internal_gap or tail_missing > maximum_rounding_pad: raise ValueError( "Lazy motion reference %s yielded only %d of %d requested " "audio samples for video window %d:%d." % (path, copied, requested, int(source_start), int(source_end))) _LOG.info( "h3_lazy_motion: tail padded %d zero audio samples while " "converting %.6g fps video to a 24 fps target window", tail_missing, float(descriptor.get("source_fps", FPS))) return { "waveform": torch.from_numpy(target).unsqueeze(0), "sample_rate": sample_rate, } def _decode_lazy_motion_source_audio( descriptor: dict[str, Any]) -> dict[str, Any]: """Decode the complete post-skip audio track on the 24 fps video clock.""" frame_count = int(descriptor.get("frame_count", 0)) if frame_count < 1: raise ValueError( "Lazy motion source audio requires a known post-skip video " "duration.") audio = _decode_lazy_motion_audio(descriptor, 0, frame_count) if audio is None: raise ValueError( "Lazy motion source video contains no embedded audio stream.") return audio def _native_video_from_path(path: str) -> Any: try: from comfy_api.latest import InputImpl except ImportError as exc: raise RuntimeError( "MiniMax H3 Lazy Motion AV Loader requires ComfyUI's native " "VIDEO API.") from exc return InputImpl.VideoFromFile(path) def _is_source_timeline(value: Any) -> bool: return ( isinstance(value, dict) and value.get("version") == SOURCE_TIMELINE_VERSION and value.get("kind") == "source_timeline" and isinstance(value.get("extent"), dict) and isinstance(value.get("audio"), dict) and isinstance(value.get("fingerprints"), dict) ) def _validate_source_timeline( timeline: Any, *, require_runtime: bool = False ) -> dict[str, Any]: if not _is_source_timeline(timeline): raise ValueError( "H3 Source Timeline must come from the MiniMax H3 Source " "Timeline node.") video = timeline.get("video") if video is not None and not _is_lazy_motion_descriptor(video): raise ValueError( "H3 Source Timeline video descriptor is invalid or obsolete.") extent = timeline["extent"] frame_count = int(extent.get("frame_count", 0)) duration = float(extent.get("duration_seconds", 0.0)) if frame_count < 1 or not math.isfinite(duration) or duration <= 0.0: raise ValueError("H3 Source Timeline has no usable 24 fps extent.") audio = timeline["audio"] audio_kind = str(audio.get("kind") or "") if audio_kind not in ("none", "embedded", "external_path", "deferred_tensor"): raise ValueError( "H3 Source Timeline has unknown audio kind %r." % audio_kind) if audio_kind in ("embedded", "external_path"): if not isinstance(audio.get("path"), str): raise ValueError( "H3 Source Timeline path-backed audio has no path.") if require_runtime: _resolved_media_path( audio["path"], "H3 Source Timeline audio recovery path") elif audio_kind == "deferred_tensor" and require_runtime: _validate_audio( audio.get("value"), "H3 Source Timeline deferred audio") if video is not None and require_runtime: _resolved_media_path( video["path"], "H3 Source Timeline video recovery path") fingerprints = timeline["fingerprints"] for key in ("video", "audio", "timeline"): value = str(fingerprints.get(key) or "") if key == "video" and video is None: if value: raise ValueError( "Audio-only H3 Source Timeline has a video fingerprint.") continue if key == "audio" and audio_kind == "none": if value: raise ValueError( "Silent H3 Source Timeline has an audio fingerprint.") continue if len(value) != 64: raise ValueError( "H3 Source Timeline %s fingerprint is invalid." % key) tracks = timeline.get("audio_tracks") if tracks is not None: if (not isinstance(tracks, dict) or tracks.get("version") != "h3_audio_tracks_v1" or tracks.get("mix_mode") not in ("full_mix", "stems") or not any(tracks.get(key) for key in ("vocals", "instrumental"))): raise ValueError("H3 Audio Tracks has an invalid track group.") for role in ("vocals", "instrumental"): track = tracks.get(role) if track is None: continue if not isinstance(track, dict) or "audio_tracks" in track: raise ValueError("H3 Audio Tracks cannot contain nested groups.") _validate_source_timeline(track, require_runtime=require_runtime) if track.get("video") is not None or track["audio"]["kind"] == "none": raise ValueError("H3 Audio Tracks %s must be audio-only." % role) if int(track["extent"]["frame_count"]) != frame_count: raise ValueError( "H3 Audio Tracks %s must have the same duration as the " "full mix, including leading silence and instrumental " "gaps. Tracks must share one timeline; they are not " "stretched or concatenated." % role) return timeline def _make_audio_tracks(full_mix=None, vocals=None, instrumental=None): """Group aligned audio-only timelines; only vocals drive a locked target.""" supplied = {key: value for key, value in ( ("full_mix", full_mix), ("vocals", vocals), ("instrumental", instrumental)) if value is not None} if not supplied: raise ValueError("H3 Audio Tracks needs a full mix, vocals, or instrumental.") for role, track in supplied.items(): _validate_source_timeline(track, require_runtime=True) if (track.get("video") is not None or "audio_tracks" in track or track["audio"]["kind"] == "none"): raise ValueError("H3 Audio Tracks %s must be a single audio track." % role) frames = {int(track["extent"]["frame_count"]) for track in supplied.values()} if len(frames) != 1: raise ValueError( "H3 Audio Tracks must have the same duration, including silence. " "Export aligned full-length stems; do not trim vocal gaps.") if full_mix is not None and len(supplied) == 1: return full_mix # Preserve the exact legacy descriptor and identity. gain = 1.0 if full_mix is None: # Decode once only when a mix must be synthesized. Cache a seekable mix # so Carousel -> Plan remains recoverable without another source wire. # Run Manager can archive it alongside the original stems. values = [_source_timeline_source_audio(track) for track in supplied.values()] sample_rate = max(int(value["sample_rate"]) for value in values) channels = max(int(_audio_waveform_3d(value, "Audio stem")[0].shape[1]) for value in values) samples = sample_boundary_from_frames(next(iter(frames)), sample_rate) parts = [_resample_audio_exact(value, sample_rate, samples, channels, "H3 Audio Tracks stem")["waveform"] for value in values] waveform = sum(parts[1:], parts[0].clone()) if not bool(torch.isfinite(waveform).all()): raise ValueError("H3 Audio Tracks contains non-finite audio samples.") peak = float(waveform.abs().max()) if peak > 1.0: gain = 1.0 / peak waveform *= gain mixed_audio = {"waveform": waveform, "sample_rate": sample_rate} mixed_path = os.path.join( folder_paths.get_temp_directory(), "h3_audio_track_mixes", "%s.wav" % _audio_fingerprint(mixed_audio)) if not os.path.isfile(mixed_path): _atomic_float_wav(mixed_audio, mixed_path) full_mix = _make_source_timeline(audio_path=mixed_path) mix_mode = "stems" else: mix_mode = "full_mix" result = dict(full_mix) result["audio_tracks"] = { "version": "h3_audio_tracks_v1", "mix_mode": mix_mode, "mix_gain": gain, "vocals": vocals, "instrumental": instrumental, } result["fingerprints"] = dict(full_mix["fingerprints"]) result["fingerprints"]["timeline"] = _fingerprint({ "version": "h3_audio_tracks_v1", "delivery": full_mix["fingerprints"]["timeline"], "vocals": vocals["fingerprints"]["timeline"] if vocals else "", "instrumental": (instrumental["fingerprints"]["timeline"] if instrumental else ""), "mix_mode": mix_mode, "mix_gain": gain, }) return _validate_source_timeline(result, require_runtime=True) def _source_timeline_generation_track(timeline, *, lip_sync=False): source = _validate_source_timeline(timeline) tracks = source.get("audio_tracks") or {} if lip_sync and tracks: if tracks.get("vocals") is None: raise ValueError( "Lip-sync is enabled, but this audio group has no vocal track. " "Attach aligned vocals or turn Lip-sync off for this scene. " "Instrumental audio is never substituted for missing vocals.") return tracks["vocals"] return source def _source_timeline_path_hash( path: str, cache: dict[str, str]) -> str: resolved = os.path.realpath(path) if resolved not in cache: cache[resolved] = _file_sha256(resolved) return cache[resolved] def _make_source_timeline( video_path: Any = "", *, source_video: Any = None, audio_path: Any = "", source_audio: Any = None, embedded_audio: str = "auto", skip_first_frames: Any = 0, source_route: str = "") -> dict[str, Any]: """Build one intent-free, file-backed source timeline descriptor.""" embedded_mode = str(embedded_audio or "auto").strip().lower() if embedded_mode not in ("auto", "required", "ignore"): raise ValueError( "Source Timeline embedded_audio must be auto, required, or " "ignore.") has_audio_path = bool(str(audio_path or "").strip()) if has_audio_path and source_audio is not None: raise ValueError( "Source Timeline accepts one external audio source: connect " "source_audio or provide audio_path, not both.") wants_video = source_video is not None or bool(str(video_path or "").strip()) video = None route = str(source_route or "").strip() if wants_video: resolved_video, detected_route = _lazy_motion_media_path( video_path, source_video) route = route or detected_route video = _lazy_motion_descriptor_with_skip( _probe_lazy_motion_path(resolved_video, False), skip_first_frames) video["source_route"] = route elif int(skip_first_frames) != 0: raise ValueError( "Source Timeline skip_first_frames needs a video clock.") hashes: dict[str, str] = {} if video is not None: video_file_hash = _source_timeline_path_hash(video["path"], hashes) video["file_sha256"] = video_file_hash video_hash = _fingerprint({ "source_timeline": SOURCE_TIMELINE_VERSION, "file_sha256": video_file_hash, "video_stream": 0, "source_rate": [ int(video["source_rate_numerator"]), int(video["source_rate_denominator"]), ], "skip_first_frames": int(video["skip_first_frames"]), }) origin_fps = float(video["source_fps"]) origin_skip = int(video["skip_first_frames"]) origin_seconds = float(video["skip_seconds"]) else: video_hash = "" origin_fps = float(FPS) origin_skip = 0 origin_seconds = 0.0 audio: dict[str, Any] if source_audio is not None: waveform, sample_rate = _validate_audio( source_audio, "H3 Source Timeline external audio") channels = 1 if waveform.ndim == 1 else int(waveform.shape[-2]) if channels not in (1, 2): raise ValueError( "H3 Source Timeline external audio must be mono or stereo.") content_hash = _audio_fingerprint(source_audio) audio = { "kind": "deferred_tensor", "value": source_audio, "sample_rate": sample_rate, "channels": channels, "duration_seconds": ( int(waveform.shape[-1]) / float(sample_rate)), "timeline_offset_seconds": 0.0, "content_sha256": content_hash, "aligned_to_timeline_origin": True, } audio_hash = _fingerprint({ "source_timeline": SOURCE_TIMELINE_VERSION, "kind": "deferred_tensor", "content_sha256": content_hash, "timeline_offset_seconds": 0.0, }) elif has_audio_path: inspected = _probe_audio_path( str(audio_path), "Source Timeline external audio path", True) assert inspected is not None audio_file_hash = _source_timeline_path_hash( inspected["path"], hashes) audio = { "kind": "external_path", **inspected, "file_sha256": audio_file_hash, "timeline_offset_seconds": origin_seconds, "origin_source_fps": origin_fps, "origin_skip_frames": origin_skip, } audio_hash = _fingerprint({ "source_timeline": SOURCE_TIMELINE_VERSION, "kind": "external_path", "file_sha256": audio_file_hash, "audio_stream": int(inspected["stream_index"]), "timeline_offset_seconds": origin_seconds, }) elif video is not None and embedded_mode != "ignore": inspected = _probe_audio_path( video["path"], "Source Timeline embedded audio", embedded_mode == "required") if inspected is None: audio = {"kind": "none"} audio_hash = "" else: audio_info = { "sample_rate": int(inspected["sample_rate"]), "channels": int(inspected["channels"]), "start_seconds": float(inspected["start_seconds"]), } video["audio"] = audio_info audio = { "kind": "embedded", **inspected, "file_sha256": video["file_sha256"], "timeline_offset_seconds": origin_seconds, "origin_source_fps": origin_fps, "origin_skip_frames": origin_skip, } audio_hash = _fingerprint({ "source_timeline": SOURCE_TIMELINE_VERSION, "kind": "embedded", "file_sha256": video["file_sha256"], "audio_stream": int(inspected["stream_index"]), "skip_first_frames": origin_skip, }) else: audio = {"kind": "none"} audio_hash = "" if video is not None: frame_count = int(video.get("frame_count", 0)) duration_seconds = frame_count / float(FPS) elif audio["kind"] != "none": available_duration = max( 0.0, float(audio.get("duration_seconds", 0.0)) - float(audio.get("timeline_offset_seconds", 0.0))) frame_count = int(math.floor(available_duration * FPS + 1e-9)) duration_seconds = frame_count / float(FPS) else: raise ValueError( "Source Timeline needs a video, an audio path, or source_audio.") if frame_count < 1: raise ValueError( "Source Timeline media has no usable duration after its origin.") timeline_hash = _fingerprint({ "version": SOURCE_TIMELINE_VERSION, "video": video_hash, "audio": audio_hash, "origin": { "source_fps": origin_fps, "skip_first_frames": origin_skip, }, "extent_frames": frame_count, }) timeline = { "version": SOURCE_TIMELINE_VERSION, "kind": "source_timeline", "fps": FPS, "origin": { "source_fps": origin_fps, "skip_first_frames": origin_skip, "skip_seconds": origin_seconds, }, "extent": { "frame_count": frame_count, "duration_seconds": duration_seconds, }, "video": video, "audio": audio, "fingerprints": { "video": video_hash, "audio": audio_hash, "timeline": timeline_hash, }, "recovery": { "video_path": video["path"] if video is not None else "", "audio_path": ( str(audio.get("path") or "") if audio["kind"] in ("embedded", "external_path") else ""), "deferred_audio_requires_materialization": ( audio["kind"] == "deferred_tensor"), }, } return _validate_source_timeline(timeline, require_runtime=True) def _source_timeline_recovery_record(timeline: Any) -> dict[str, Any]: """Return a JSON-safe descriptor suitable for run/checkpoint metadata.""" source = _validate_source_timeline(timeline) record = { "version": source["version"], "kind": source["kind"], "fps": int(source["fps"]), "origin": dict(source["origin"]), "extent": dict(source["extent"]), "video": dict(source["video"]) if source.get("video") else None, "audio": { key: value for key, value in source["audio"].items() if key != "value" }, "fingerprints": dict(source["fingerprints"]), "recovery": dict(source.get("recovery") or {}), } if record["audio"].get("kind") == "deferred_tensor": record["audio"]["requires_materialization"] = True if source.get("audio_tracks") is not None: record["audio_tracks"] = { **source["audio_tracks"], **{role: _source_timeline_recovery_record(track) for role in ("vocals", "instrumental") if (track := source["audio_tracks"].get(role)) is not None}, } # Fail here, close to persistence, rather than writing an unusable run. _canonical_json(record) return record def _source_timeline_verify_recovery_file( descriptor: dict[str, Any], label: str) -> None: path = str(descriptor.get("path") or "") if not path or not os.path.isfile(path): return expected = str(descriptor.get("file_sha256") or "") if expected and _file_sha256(path) != expected: raise ValueError( "%s exists but its contents do not match the saved Source " "Timeline fingerprint: %s" % (label, path)) def _source_timeline_from_recovery( record: Any, *, video_path: Any = "", audio_path: Any = "", deferred_audio: Any = None, track_paths=None, deferred_tracks=None ) -> dict[str, Any]: """Restore or relink a persisted timeline without changing its identity.""" try: restored = json.loads(_canonical_json(record)) except (TypeError, ValueError) as exc: raise ValueError( "Saved H3 Source Timeline descriptor is not JSON-safe.") from exc _validate_source_timeline(restored) video = restored.get("video") audio = restored["audio"] if str(video_path or "").strip(): if video is None: raise ValueError("Cannot relink video on an audio-only timeline.") resolved = _resolved_media_path( video_path, "H3 Source Timeline replacement video") if _file_sha256(resolved) != str(video.get("file_sha256") or ""): raise ValueError( "Replacement video does not match the saved Source Timeline.") video["path"] = resolved restored["recovery"]["video_path"] = resolved if audio.get("kind") == "embedded": audio["path"] = resolved restored["recovery"]["audio_path"] = resolved if str(audio_path or "").strip(): if audio.get("kind") != "external_path": raise ValueError( "audio_path relinking is only valid for external-path audio.") resolved = _resolved_media_path( audio_path, "H3 Source Timeline replacement audio") if _file_sha256(resolved) != str(audio.get("file_sha256") or ""): raise ValueError( "Replacement audio does not match the saved Source Timeline.") audio["path"] = resolved restored["recovery"]["audio_path"] = resolved if audio.get("kind") == "deferred_tensor" and deferred_audio is not None: _validate_audio(deferred_audio, "Recovered H3 Source Timeline audio") expected = str(audio.get("content_sha256") or "") if _audio_fingerprint(deferred_audio) != expected: raise ValueError( "Recovered audio tensor does not match the saved Source " "Timeline fingerprint.") audio["value"] = deferred_audio if video is not None: _source_timeline_verify_recovery_file( video, "H3 Source Timeline video recovery path") if audio.get("kind") == "external_path": _source_timeline_verify_recovery_file( audio, "H3 Source Timeline audio recovery path") if restored.get("audio_tracks") is not None: tracks = restored["audio_tracks"] for role in ("vocals", "instrumental"): if tracks.get(role) is not None: tracks[role] = _source_timeline_from_recovery( tracks[role], audio_path=(track_paths or {}).get(role, ""), deferred_audio=(deferred_tracks or {}).get(role)) return _validate_source_timeline(restored) def _source_timeline_audio_descriptor( timeline: dict[str, Any]) -> dict[str, Any] | None: source = _validate_source_timeline(timeline) audio = source["audio"] kind = audio["kind"] if kind == "none" or kind == "deferred_tensor": return None if kind == "embedded": descriptor = dict(source["video"]) descriptor["audio"] = { "sample_rate": int(audio["sample_rate"]), "channels": int(audio["channels"]), "start_seconds": float(audio["start_seconds"]), } return descriptor return { "version": LAZY_MOTION_SOURCE_VERSION, "kind": "lazy_motion_path", "path": audio["path"], "fps": FPS, "source_fps": float(audio.get("origin_source_fps", FPS)), "skip_first_frames": int(audio.get("origin_skip_frames", 0)), "video_start_seconds": float(audio.get("start_seconds", 0.0)), "frame_count": int(source["extent"]["frame_count"]), "audio": { "sample_rate": int(audio["sample_rate"]), "channels": int(audio["channels"]), "start_seconds": float(audio.get("start_seconds", 0.0)), }, } def _source_timeline_scene_video( timeline: Any, source_start: int, source_end: int) -> Any: source = _validate_source_timeline(timeline, require_runtime=True) if source.get("video") is None: return None start = int(source_start) end = int(source_end) if start < 0 or end > int(source["extent"]["frame_count"]): raise ValueError( "H3 Source Timeline video window %d:%d exceeds 0:%d." % (start, end, int(source["extent"]["frame_count"]))) return _decode_lazy_motion_video(source["video"], start, end) def _source_timeline_scene_audio( timeline: Any, source_start: int, source_end: int) -> Any: source = _validate_source_timeline(timeline, require_runtime=True) start = int(source_start) end = int(source_end) if start < 0 or end <= start or end > int( source["extent"]["frame_count"]): raise ValueError( "H3 Source Timeline audio window %d:%d exceeds 0:%d." % (start, end, int(source["extent"]["frame_count"]))) audio = source["audio"] if audio["kind"] == "none": return None if audio["kind"] == "deferred_tensor": return _slice_audio( audio["value"], start / float(FPS), (end - start) / float(FPS)) descriptor = _source_timeline_audio_descriptor(source) return _decode_lazy_motion_audio(descriptor, start, end) def _source_timeline_source_audio(timeline: Any) -> Any: source = _validate_source_timeline(timeline, require_runtime=True) if source["audio"]["kind"] == "none": return None return _source_timeline_scene_audio( source, 0, int(source["extent"]["frame_count"])) def _source_timeline_to_lazy_motion_descriptor( timeline: Any) -> dict[str, Any] | None: source = _validate_source_timeline(timeline) if source.get("video") is None: return None descriptor = dict(source["video"]) if source["audio"]["kind"] != "embedded": descriptor["audio"] = None return descriptor def _source_timeline_from_lazy_motion_descriptor( descriptor: Any, source_audio: Any = None) -> dict[str, Any]: if not _is_lazy_motion_descriptor(descriptor): raise ValueError( "Legacy motion source is not a valid lazy descriptor.") return _make_source_timeline( descriptor["path"], source_audio=source_audio, embedded_audio=( "required" if descriptor.get("audio") is not None else "ignore"), skip_first_frames=int(descriptor.get("skip_first_frames", 0)), source_route=str(descriptor.get("source_route") or "legacy adapter")) def _legacy_media_to_source_timeline( video_path: Any = "", *, source_video: Any = None, source_audio: Any = None, skip_first_frames: Any = 0, use_embedded_audio: bool = True) -> dict[str, Any]: return _make_source_timeline( video_path, source_video=source_video, source_audio=source_audio, embedded_audio="required" if use_embedded_audio else "ignore", skip_first_frames=skip_first_frames, source_route="legacy adapter") def _source_timeline_with_motion_short_edge( timeline: Any, short_edge: Any) -> dict[str, Any]: """Attach a decode-only resize choice without changing media identity.""" source = _validate_source_timeline(timeline) if source.get("video") is None: raise ValueError( "Tagged motion reference requires video on H3 Source Timeline.") resolved = str(short_edge or "384") if resolved not in MOTION_REFERENCE_SHORT_EDGES: raise ValueError("Unknown motion-reference short edge %r." % short_edge) prepared = dict(source) prepared["video"] = dict(source["video"]) prepared["video"]["motion_short_edge"] = resolved prepared["audio"] = dict(source["audio"]) prepared["fingerprints"] = dict(source["fingerprints"]) prepared["recovery"] = dict(source.get("recovery") or {}) return _validate_source_timeline(prepared) def _source_timeline_available_audio_frames(timeline: Any) -> int: source = _validate_source_timeline(timeline) audio = source["audio"] if audio["kind"] == "none": return 0 available_seconds = max( 0.0, float(audio.get("duration_seconds", 0.0)) - float(audio.get("timeline_offset_seconds", 0.0))) return int(math.floor(available_seconds * FPS + 1e-9)) def _materialize_source_timeline_audio( timeline: Any, plan: dict[str, Any]) -> dict[str, Any]: timeline = _validate_source_timeline(timeline, require_runtime=True) if (timeline.get("audio_tracks") and _audio_policy_requires_source(plan) and _source_timeline_available_audio_frames(timeline) < int(plan["total_delivered_frames"])): raise ValueError( "H3 Audio Tracks is too short for this Plan. Supply aligned " "full-length tracks; grouped stems are never padded separately.") source = _materialize_source_timeline_primary_audio(timeline, plan) if source.get("audio_tracks") is not None: source = dict(source) source["audio_tracks"] = { **source["audio_tracks"], **{role: _materialize_source_timeline_primary_audio(track, plan) for role in ("vocals", "instrumental") if (track := source["audio_tracks"].get(role)) is not None}, } return _validate_source_timeline(source, require_runtime=True) def _materialize_source_timeline_primary_audio( timeline: Any, plan: dict[str, Any]) -> dict[str, Any]: """Persist a deferred AUDIO once so recursive state remains path-backed.""" source = _validate_source_timeline(timeline, require_runtime=True) audio = source["audio"] if audio["kind"] != "deferred_tensor": return source value = audio["value"] waveform, sample_rate = _validate_audio( value, "H3 Source Timeline deferred audio") required_samples = int(round( int(plan["total_delivered_frames"]) / float(FPS) * sample_rate)) materialized = value silent_padding = False if (_audio_policy_requires_source(plan) and int(waveform.shape[-1]) < required_samples): if not _audio_is_silent(waveform): raise ValueError( "H3 Source Timeline audio is too short: it contains %d " "samples at %d Hz, but this plan requires at least %d " "samples for %d delivered frames. Only silent placeholder " "audio is automatically padded." % (int(waveform.shape[-1]), sample_rate, required_samples, int(plan["total_delivered_frames"]))) materialized = _pad_audio_to_samples( value, required_samples, "H3 Source Timeline silent placeholder audio") silent_padding = True content_hash = str(audio.get("content_sha256") or _audio_fingerprint(value)) directory = os.path.join(_run_dir(plan), "source_timeline") path = os.path.join(directory, "source_audio.%s.wav" % content_hash[:16]) if not os.path.isfile(path): _atomic_float_wav(materialized, path) inspected = _probe_audio_path( path, "Materialized H3 Source Timeline audio", True) assert inspected is not None prepared = dict(source) prepared["video"] = (dict(source["video"]) if source.get("video") is not None else None) prepared["audio"] = { "kind": "external_path", **inspected, "file_sha256": _file_sha256(path), "timeline_offset_seconds": 0.0, "origin_source_fps": float(FPS), "origin_skip_frames": 0, "content_sha256": content_hash, "materialized_from_tensor": True, } prepared["fingerprints"] = dict(source["fingerprints"]) prepared["recovery"] = dict(source.get("recovery") or {}) prepared["recovery"].update({ "audio_path": path, "deferred_audio_requires_materialization": False, "materialized_audio": True, }) if silent_padding: prepared["recovery"]["silent_padding_frames"] = int( plan["total_delivered_frames"]) if prepared.get("video") is None: prepared["extent"] = { "frame_count": int(plan["total_delivered_frames"]), "duration_seconds": ( int(plan["total_delivered_frames"]) / float(FPS)), } return _validate_source_timeline(prepared, require_runtime=True) def _path_is_inside(directory: str, path: str) -> bool: try: return os.path.commonpath(( os.path.realpath(directory), os.path.realpath(path))) == ( os.path.realpath(directory)) except ValueError: return False def _archive_source_timeline_path( source_path: str, expected_hash: str, destination_dir: str) -> str: source_path = _resolved_media_path( source_path, "H3 Source Timeline archive source") expected_hash = str(expected_hash or "") actual_hash = _file_sha256(source_path) if expected_hash and actual_hash != expected_hash: raise ValueError( "H3 Source Timeline media changed before Run Manager archived it.") os.makedirs(destination_dir, exist_ok=True) basename = _safe_name( os.path.splitext(os.path.basename(source_path))[0], "source") suffix = os.path.splitext(source_path)[1][:24] destination = os.path.join( destination_dir, "%s_%s%s" % (actual_hash, basename, suffix)) if os.path.isfile(destination): if _file_sha256(destination) != actual_hash: raise ValueError( "H3 Source Timeline archive contains a corrupt existing file: " "%s" % destination) return destination temporary = "%s.%s.tmp" % (destination, uuid.uuid4().hex) try: with open(source_path, "rb") as source_handle, open( temporary, "xb") as target_handle: shutil.copyfileobj(source_handle, target_handle, 1024 * 1024) if _file_sha256(temporary) != actual_hash: raise ValueError( "H3 Source Timeline archive copy failed verification.") os.replace(temporary, destination) finally: _safe_unlink(temporary) return destination def _archive_source_timeline_media( timeline: Any, plan: dict[str, Any], *, archive_audio: bool, archive_video: bool) -> dict[str, Any]: """Apply Run Manager archive policy without decoding lazy media.""" source = _validate_source_timeline(timeline, require_runtime=True) prepared = dict(source) prepared["video"] = (dict(source["video"]) if source.get("video") is not None else None) prepared["audio"] = dict(source["audio"]) prepared["fingerprints"] = dict(source["fingerprints"]) prepared["recovery"] = dict(source.get("recovery") or {}) run_dir = _run_dir(plan) video = prepared.get("video") audio = prepared["audio"] if video is not None and bool(archive_video): original = video["path"] archived = _archive_source_timeline_path( original, str(video.get("file_sha256") or ""), os.path.join(run_dir, "references", "video")) prepared["recovery"]["original_video_path"] = original video["path"] = archived prepared["recovery"]["video_path"] = archived prepared["recovery"]["video_archived"] = True if audio["kind"] == "embedded": prepared["recovery"]["original_audio_path"] = audio["path"] audio["path"] = archived prepared["recovery"]["audio_path"] = archived prepared["recovery"]["audio_archived_with_video"] = True if audio["kind"] == "external_path" and bool(archive_audio): original = audio["path"] if not _path_is_inside(run_dir, original): archived = _archive_source_timeline_path( original, str(audio.get("file_sha256") or ""), os.path.join(run_dir, "references", "audio")) prepared["recovery"]["original_audio_path"] = original audio["path"] = archived prepared["recovery"]["audio_path"] = archived prepared["recovery"]["audio_archived"] = True if (audio["kind"] == "embedded" and bool(archive_audio) and not bool(archive_video)): prepared["recovery"]["audio_archive_requires_video"] = True if source.get("audio_tracks") is not None: prepared["audio_tracks"] = { **source["audio_tracks"], **{role: _archive_source_timeline_media( track, plan, archive_audio=archive_audio, archive_video=False) for role in ("vocals", "instrumental") if (track := source["audio_tracks"].get(role)) is not None}, } return _validate_source_timeline(prepared, require_runtime=True) def _validate_source_timeline_hash( compatibility: dict[str, Any], timeline: Any, usage: str) -> None: source = _validate_source_timeline(timeline, require_runtime=True) expected_timeline = str( compatibility.get("source_timeline_fingerprint") or "") actual_timeline = str(source["fingerprints"]["timeline"]) expected_audio = str(compatibility.get("source_audio_hash") or "") legacy_audio = str( (source.get("recovery") or {}).get( "legacy_loop_start_source_audio_hash") or "") legacy_recovery = bool( source.get("video") is None and expected_audio and expected_audio != "none" and legacy_audio == expected_audio and str(source["audio"].get("content_sha256") or "") == expected_audio) if (expected_timeline and expected_timeline != actual_timeline) or ( not expected_timeline and not legacy_recovery): raise ValueError( "%s received a different H3 Source Timeline than Loop Start." % usage) if expected_audio and expected_audio != "none": actual_audio = str(source["fingerprints"].get("audio") or "none") content_audio = str(source["audio"].get("content_sha256") or "") if (not expected_audio or expected_audio not in (actual_audio, content_audio)): raise ValueError( "%s received different Source Timeline audio than Loop " "Start." % usage) def _source_timeline_recovers_legacy_audio( compatibility: dict[str, Any], timeline: Any) -> bool: """Whether a timeline is the saved form of Loop Start's legacy AUDIO.""" try: source = _validate_source_timeline(timeline, require_runtime=True) except (OSError, TypeError, ValueError): return False expected = str(compatibility.get("source_audio_hash") or "") return bool( source.get("video") is None and expected and expected != "none" and str((source.get("recovery") or {}).get( "legacy_loop_start_source_audio_hash") or "") == expected and str(source["audio"].get("content_sha256") or "") == expected) def _source_timeline_from_metadata(value: Any) -> dict[str, Any] | None: if not isinstance(value, dict): return None record = value.get("source_timeline") if record is None and isinstance(value.get("plan"), dict): record = value["plan"].get("source_timeline") if record is None: return None return _source_timeline_from_recovery(record) def _scheduled_video_reference_slice( entry: dict[str, Any], state: Any, scene: int, scene_count: int, length: int) -> tuple[Any, Any, str]: """Resolve one active video ref without silently restarting its timeline.""" timeline_mode = str( entry.get("timeline_mode") or "restart_each_scene") video = entry["value"] audio = entry.get("audio") if timeline_mode == "restart_each_scene": if _is_source_timeline(video): source_start = 0 source_end = int(video["extent"]["frame_count"]) decoded_video = _source_timeline_scene_video( video, source_start, source_end) decoded_audio = ( _source_timeline_scene_audio( video, source_start, source_end) if audio is not None else None) return decoded_video, decoded_audio, ( "@%s Source Timeline restart frames %d:%d" % (entry.get("tag", "video"), source_start, source_end)) if _is_lazy_motion_descriptor(video): source_start = 0 source_end = int(video.get("frame_count", 0)) if source_end < 1: raise ValueError( "Scheduled video @%s has no usable reference frames." % entry.get("tag", "video")) decoded_video = _decode_lazy_motion_video( video, source_start, source_end) decoded_audio = _decode_lazy_motion_audio( video, source_start, source_end) return decoded_video, decoded_audio, ( "@%s lazy restart frames %d:%d" % (entry.get("tag", "video"), source_start, source_end)) return video, audio, "" if timeline_mode != "sequential": raise ValueError( "Scheduled video @%s has unknown timeline mode %r." % (entry.get("tag", "video"), timeline_mode)) if not isinstance(state, dict) or not isinstance(state.get("plan"), dict): raise ValueError( "Sequential scheduled video @%s requires Current Shot state on " "Scheduled Ref2VA's state input." % entry.get("tag", "video")) if int(state.get("index", -1)) != int(scene): raise ValueError( "Scheduled Ref2VA received scene %d but Current Shot state is at " "scene %s." % (int(scene), state.get("index"))) shots = state["plan"].get("shots") if not isinstance(shots, list) or len(shots) != int(scene_count): raise ValueError( "Sequential scheduled video requires state from the same %d-scene " "Plan connected to Scheduled Ref2VA." % int(scene_count)) current = shots[int(scene) - 1] if int(current.get("raw_frames", -1)) != int(length): raise ValueError( "Scheduled Ref2VA length %d does not match scene %d's %s raw " "frames in Current Shot state." % (int(length), int(scene), current.get("raw_frames"))) ranges = entry.get("ranges") or () if entry.get("activation") == "prompt": origin_scene = next(( index for index, shot in enumerate(shots, 1) if set(_REFERENCE_ALIAS_RE.findall(_prompt_text( shot.get("prompt", ""), "Scene %d prompt" % index ))).intersection(_reference_entry_tags(entry)) ), int(scene)) else: origin_scene = int(ranges[0][0]) if ranges else 1 if origin_scene < 1 or origin_scene > len(shots): raise ValueError( "Sequential scheduled video @%s has an invalid first active " "scene %d." % (entry.get("tag", "video"), origin_scene)) origin_start = int(shots[origin_scene - 1]["generation_start_frame"]) current_start = int(current["generation_start_frame"]) source_start = current_start - origin_start source_length = int(length) audio_source_start = source_start audio_source_length = source_length detail_mode = "" compatibility = state["plan"].get("compatibility") if not isinstance(compatibility, dict): compatibility = {} continuation_mode = current.get( "continuation_mode", compatibility.get("continuation_mode", "guide")) if (entry.get("semantic_role") == "motion" and continuation_mode in MASKED_CONTINUATION_MODES): # Native Ref2VA video banks carry no target-frame coordinates. A # masked continuation already owns the repeated prefix, so including # those reference frames makes H3 replay them once the denoisable tail # begins (one context span late). Advance only the motion video on # the delivered timeline. Its paired soundtrack still describes the # complete raw target, including the repeated prefix, and therefore # must remain on the raw generation window. Applying the delivered # video window to audio shifts source-driven performance by exactly # one context span on every continuation scene. delivered = current.get("delivered_frames") if delivered is None: raise ValueError( "Sequential motion reference @%s requires delivered-frame " "timing from Current Shot state." % entry.get("tag", "video")) preceding = shots[origin_scene - 1:int(scene) - 1] if any(shot.get("delivered_frames") is None for shot in preceding): raise ValueError( "Sequential motion reference @%s requires delivered-frame " "timing for every scene since its first activation." % entry.get("tag", "video")) source_start = sum(int(shot["delivered_frames"]) for shot in preceding) source_length = int(delivered) if source_length < 1: raise ValueError( "Sequential motion reference @%s resolved an empty delivered " "window for scene %d." % (entry.get("tag", "video"), int(scene))) detail_mode = " delivered" source_end = source_start + source_length audio_source_end = audio_source_start + audio_source_length if source_start < 0 or audio_source_start < 0: raise ValueError( "Sequential scheduled video @%s resolved a negative source " "window for scene %d." % (entry.get("tag", "video"), int(scene))) if _is_source_timeline(video): available = int(video["extent"]["frame_count"]) elif _is_lazy_motion_descriptor(video): available = int(video.get("frame_count", 0)) else: available = int(video.shape[0]) if available > 0 and source_end > available: raise ValueError( "Sequential scheduled video @%s needs source frames %d:%d for " "scene %d, but the 24 fps reference contains only %d frames. " "Supply a longer reference, shorten the Plan, or use " "restart_each_scene." % (entry.get("tag", "video"), source_start, source_end, int(scene), available)) if _is_source_timeline(video): sliced_video = _source_timeline_scene_video( video, source_start, source_end) sliced_audio = ( _source_timeline_scene_audio( video, audio_source_start, audio_source_end) if audio is not None else None) elif _is_lazy_motion_descriptor(video): sliced_video = _decode_lazy_motion_video( video, source_start, source_end) sliced_audio = _decode_lazy_motion_audio( video, audio_source_start, audio_source_end) else: sliced_video = video[source_start:source_end] sliced_audio = None if (audio is not None and not _is_lazy_motion_descriptor(video) and not _is_source_timeline(video)): waveform, sample_rate = _validate_audio( audio, "Sequential scheduled video @%s audio" % entry.get( "tag", "video")) sample_start = int(round( audio_source_start / float(FPS) * sample_rate)) sample_end = int(round( audio_source_end / float(FPS) * sample_rate)) available_samples = int(waveform.shape[-1]) if sample_end > available_samples: raise ValueError( "Sequential scheduled video @%s needs paired-audio samples " "%d:%d for scene %d, but the soundtrack contains only %d " "samples at %d Hz." % (entry.get("tag", "video"), sample_start, sample_end, int(scene), available_samples, sample_rate)) sliced_audio = { "waveform": waveform[..., sample_start:sample_end], "sample_rate": sample_rate, } if (sliced_audio is not None and (audio_source_start != source_start or audio_source_end != source_end)): detail = ( "@%s sequential%s video frames %d:%d; paired audio raw frames " "%d:%d (origin scene %d)" % ( entry.get("tag", "video"), detail_mode, source_start, source_end, audio_source_start, audio_source_end, origin_scene)) else: detail = "@%s sequential%s frames %d:%d (origin scene %d)" % ( entry.get("tag", "video"), detail_mode, source_start, source_end, origin_scene) return sliced_video, sliced_audio, detail def _tagged_audio_reference_value( entry: dict[str, Any], state: Any, scene: int, scene_count: int, length: int) -> tuple[Any, str]: """Resolve a tagged standalone clip or an exact Plan-timeline slice.""" timeline_mode = str(entry.get("timeline_mode") or "standalone") audio = _project_asset_audio(entry["value"]) if timeline_mode == "standalone": return audio, "" if timeline_mode != "source_timeline": raise ValueError( "Tagged audio @%s has unknown timeline mode %r." % (entry.get("tag", "audio"), timeline_mode)) if not isinstance(state, dict) or not isinstance(state.get("plan"), dict): raise ValueError( "Tagged audio @%s uses source_timeline and requires Current Shot " "state on Tagged Ref2VA's state input. Keep the full source track " "connected to Tagged Audio Ref; do not connect " "source_audio_slice there because its fingerprint-to-Plan link " "would create a cycle." % entry.get("tag", "audio")) if int(state.get("index", -1)) != int(scene): raise ValueError( "Tagged Ref2VA received scene %d but Current Shot state is at " "scene %s." % (int(scene), state.get("index"))) plan = state["plan"] shots = plan.get("shots") if not isinstance(shots, list) or len(shots) != int(scene_count): raise ValueError( "Tagged audio @%s source_timeline requires state from the same " "%d-scene Plan connected to Tagged Ref2VA." % (entry.get("tag", "audio"), int(scene_count))) current = shots[int(scene) - 1] if int(current.get("raw_frames", -1)) != int(length): raise ValueError( "Tagged Ref2VA length %d does not match scene %d's %s raw frames " "in Current Shot state." % (int(length), int(scene), current.get("raw_frames"))) compatibility = plan.get("compatibility") if (not isinstance(compatibility, dict) or not _audio_policy_uses_source_reference(plan, current)): raise ValueError( "Tagged audio @%s source_timeline requires Source reference=on " "in this scene's effective audio policy." % entry.get("tag", "audio")) expected_hash = str(compatibility.get("source_audio_hash") or "") entry_hash = str(entry.get("content_hash") or "") if not expected_hash or expected_hash == "none": raise ValueError( "Tagged audio @%s source_timeline has no Loop Start source-audio " "fingerprint to validate." % entry.get("tag", "audio")) accepted_hashes = {expected_hash} source_timeline = state.get("source_timeline") if _is_source_timeline(source_timeline): timeline_audio_hash = str( source_timeline["fingerprints"].get("audio") or "") timeline_content_hash = str( source_timeline["audio"].get("content_sha256") or "") # Source Timeline deliberately fingerprints the media route as well # as its decoded content. A connected AUDIO tensor is registered by # Tagged Audio with the content fingerprint, while Loop Start records # the route fingerprint in the Plan. Accept the content identity only # when this runtime timeline is itself the source recorded by Plan. if expected_hash in (timeline_audio_hash, timeline_content_hash): accepted_hashes.update(filter(None, ( timeline_audio_hash, timeline_content_hash))) if not entry_hash or entry_hash not in accepted_hashes: raise ValueError( "Tagged audio @%s source_timeline received a different full " "source track than H3 Chain Loop Start. Wire the same Load Audio " "output to both nodes." % entry.get("tag", "audio")) external_lead = int(current.get("external_context_frames", 0)) if int(scene) == 1 and external_lead > 0: sliced = _slice_audio_after_external_context( audio, state.get("previous_audio"), int(current["raw_frames"]), external_lead, pad_silence=bool(compatibility.get( "source_audio_silent_padding"))) else: sliced = _slice_audio( audio, current["audio_start_seconds"], current["audio_duration_seconds"], pad_silence=bool(compatibility.get( "source_audio_silent_padding"))) if bool(entry.get("align_audio_reference")): sliced, alignment = _align_audio_reference_to_h3_grid( sliced, int(length)) else: alignment = "frame-exact" start = float(current["audio_start_seconds"]) end = start + float(current["audio_duration_seconds"]) detail = "@%s source timeline %.3f..%.3fs; %s" % ( entry.get("tag", "audio"), start, end, alignment) return sliced, detail def _active_reference_bindings( schedule: Any, scene: int, scene_count: int, compliance_mode: str = "strict", warnings: list[str] | None = None, activation_tags: set[str] | None = None) -> dict[str, Any]: scene, scene_count = int(scene), int(scene_count) mode = _reference_compliance_mode(compliance_mode) warnings = warnings if warnings is not None else [] if scene_count < 1 or scene < 1 or scene > scene_count: raise ValueError( "Scheduled Ref2VA scene index must be between 1 and %d; got %d." % (scene_count, scene)) if activation_tags is None: entries = _reference_schedule_entries(schedule) for entry in entries: try: _parse_reference_selector( entry.get("scenes", "all"), scene_count) except ValueError as exc: if mode != "disabled": raise warnings.append(str(exc)) _LOG.warning("H3 scheduled-reference warning: %s", exc) active = [ entry for entry in entries if _reference_is_active(entry, scene)] else: entries = _tagged_reference_entries(schedule) active = sorted( [entry for entry in entries if activation_tags.intersection(_reference_entry_tags(entry))], key=lambda entry: ( str(entry.get("kind") or ""), str(entry.get("tag") or ""), str(entry.get("audio_tag") or ""), )) pictures = [entry for entry in active if entry.get("kind") == "picture"] videos = [entry for entry in active if entry.get("kind") == "video"] audios = [entry for entry in active if entry.get("kind") == "audio"] unknown = [entry.get("kind") for entry in active if entry.get("kind") not in ("picture", "video", "audio")] if unknown: message = "Unknown scheduled reference kind %r." % unknown[0] if mode != "disabled": raise ValueError(message) warnings.append(message) _LOG.warning("H3 scheduled-reference warning: %s", message) if len(pictures) > 9: message = ( "Scene %d activates %d pictures; only the first 9 were kept." % (scene, len(pictures))) if mode != "disabled": raise ValueError( "Scene %d activates %d pictures; stock H3 Ref2VA supports 9." % (scene, len(pictures))) warnings.append(message) _LOG.warning("H3 scheduled-reference warning: %s", message) pictures = pictures[:9] if len(videos) > 3: message = ( "Scene %d activates %d videos; only the first 3 were kept." % (scene, len(videos))) if mode != "disabled": raise ValueError( "Scene %d activates %d videos; stock H3 Ref2VA supports 3." % (scene, len(videos))) warnings.append(message) _LOG.warning("H3 scheduled-reference warning: %s", message) videos = videos[:3] if len(audios) > 3: message = ( "Scene %d activates %d standalone audios; only the first 3 " "were kept." % (scene, len(audios))) if mode != "disabled": raise ValueError( "Scene %d activates %d standalone audios; stock H3 Ref2VA " "supports 3." % (scene, len(audios))) warnings.append(message) _LOG.warning("H3 scheduled-reference warning: %s", message) audios = audios[:3] aliases: dict[str, str] = {} presentation: list[dict[str, Any]] = [] for ordinal, entry in enumerate(pictures, 1): label = "" % ordinal aliases[entry["tag"]] = label presentation.append({ "entry": entry, "role": "picture", "tag": entry["tag"], "label": label, }) audio_ordinal = 0 for ordinal, entry in enumerate(videos, 1): if entry.get("audio") is not None: audio_ordinal += 1 audio_label = "