#!/usr/bin/env python3 """The compact policy is one wire with identical resolved Plan semantics.""" import importlib.util import json import pathlib import sys import types ROOT = pathlib.Path(__file__).resolve().parents[1] PACKAGE = "h3_chain_policy_unit" folder_paths = types.ModuleType("folder_paths") folder_paths.get_output_directory = lambda: str(ROOT) folder_paths.get_temp_directory = lambda: str(ROOT) folder_paths.get_input_directory = lambda: str(ROOT) folder_paths.get_annotated_filepath = lambda value: str(value) sys.modules["folder_paths"] = folder_paths package = types.ModuleType(PACKAGE) package.__path__ = [str(ROOT)] sys.modules[PACKAGE] = package shared_nodes = types.ModuleType(PACKAGE + ".nodes") shared_nodes.MiniMaxH3MotionContext = object shared_nodes._claim_inline_patch_ownership = lambda _conditioning=None: "test patch owner" shared_nodes._prepare_native_guide_conditioning = lambda value: value shared_nodes._resize = lambda *args: None shared_nodes._streams_from_latent = lambda *args: None sys.modules[shared_nodes.__name__] = shared_nodes spec = importlib.util.spec_from_file_location( PACKAGE + ".chain_nodes", ROOT / "chain_nodes.py") chain = importlib.util.module_from_spec(spec) sys.modules[spec.name] = chain spec.loader.exec_module(chain) PLAN_JSON = json.dumps({ "shots": [ {"id": "one", "prompt": "First scene.", "length": 73}, {"id": "two", "prompt": "Second scene.", "length": 73}, ], }) def make_plan(*, combined=None, audio_context_length=22): return chain._normalize_plan( PLAN_JSON, "compact-policy-test", 64, 64, 22, "video", "head", "disabled", "generated_audio", audio_context_length, 3.0, 8, 7, 18, "model-stack", 0, "guide", combined) node = chain.MiniMaxH3ChainPolicy() required = node.INPUT_TYPES()["required"] assert tuple(required["incoming_transition"][0]) == ( "cut", "guide", "hard_av", "soft_av") assert "audio_context_length" not in required assert list(required)[-1] == "lock_source_audio" assert required["lock_source_audio"][0] == "BOOLEAN" assert required["lock_source_audio"][1]["default"] is False assert node.DEPRECATED is True combined, status = node.build("soft_av", "source", "on", "off") assert combined["version"] == chain.CHAIN_POLICY_VERSION assert combined["audio_policy"] == chain._contract_audio_policy( "source", "on", "off") assert combined["transition_policy"] == chain._contract_transition_policy( "soft_av") assert combined["audio_context_length"] == 39 assert "Soft AV" in status assert "final=source/ref=on/carry=off" in status assert "source timeline required" in status assert "audio context automatic (39f)" in status locked, locked_status = node.build( "soft_av", "source", "on", "on", True) assert locked["audio_policy"] == chain._contract_audio_policy( "source", "off", "off", "locked") assert "final=source/ref=off/carry=off/target=locked" in locked_status assert "source timeline required" in locked_status profile_node = chain.MiniMaxH3GenerationProfile() profile_inputs = profile_node.INPUT_TYPES()["required"] assert profile_node.INPUT_TYPES()["optional"]["lip_sync_options"][0] == ( chain.LIP_SYNC_OPTIONS_TYPE) assert tuple(profile_inputs["scene_continuity"][0]) == ( "Visual continuity", "Independent scenes", "Hard picture + protected audio", "Hard picture + smooth audio") assert tuple(profile_inputs["audio_profile"][0]) == ( "Generate audio", "Generate fresh audio per scene", "Lip-sync to source audio", "Generate audio from source guide", "Use source soundtrack only", "No final audio") assert profile_inputs["scene_continuity"][1]["display_name"] == ( "Scene continuity") assert profile_inputs["audio_profile"][1]["display_name"] == "Audio profile" generated_profile, generated_profile_status = profile_node.build() assert generated_profile["audio_policy"] == chain._contract_audio_policy( "generated", "off", "on") assert generated_profile["transition_policy"] == ( chain._contract_transition_policy("guide")) assert generated_profile["audio_context_length"] == 22 assert "Generate audio" in generated_profile_status lip_sync_profile, lip_sync_status = profile_node.build( "Hard picture + smooth audio", "Lip-sync to source audio") assert lip_sync_profile["audio_policy"] == chain._contract_audio_policy( "source", "off", "off", "locked") assert lip_sync_profile["transition_policy"] == ( chain._contract_transition_policy("soft_av")) assert lip_sync_profile["audio_context_length"] == 39 assert "source timeline required" in lip_sync_status options_node = chain.MiniMaxH3LipSyncOptions() lip_options, passthrough_voice, options_status = options_node.build() assert passthrough_voice is None assert lip_options["version"] == chain.LIP_SYNC_OPTIONS_VERSION assert lip_options["preroll_seconds"] == 1.0 assert lip_options["lookahead_seconds"] == 0.2 assert "vocal stem not connected" in options_status contextual_profile, contextual_status = profile_node.build( "Hard picture + smooth audio", "Lip-sync to source audio", lip_options) assert contextual_profile["audio_policy"]["lip_sync_options"] == lip_options assert "contextual song options active" in contextual_status ignored_options = profile_node.build( "Visual continuity", "Generate audio", lip_options)[0] assert "lip_sync_options" not in ignored_options["audio_policy"] source_guide_profile = profile_node.build( "Independent scenes", "Generate audio from source guide")[0] assert source_guide_profile["audio_policy"] == chain._contract_audio_policy( "generated", "on", "off") assert source_guide_profile["audio_context_length"] == 0 advanced = chain.MiniMaxH3AdvancedPolicy() advanced_inputs = advanced.INPUT_TYPES()["required"] assert advanced_inputs["chain_policy"][0] == chain.CHAIN_POLICY_TYPE assert tuple(advanced_inputs["incoming_transition"][0]) == ( "cut", "guide", "tone_guide", "latent_guide", "detail_guide", "detail_av", "drift_av", "color_drift_av", "hard_av", "soft_av") drift, drift_status = advanced.apply(combined, "drift_av") assert drift["audio_policy"] == combined["audio_policy"] assert drift["transition_policy"] == chain._contract_transition_policy( "drift_av") assert drift["audio_context_length"] == 39 assert "advanced override" in drift_status assert "audio preserved" in drift_status locked_drift = advanced.apply(locked, "drift_av")[0] assert locked_drift["audio_policy"] == locked["audio_policy"] contextual_drift = advanced.apply(contextual_profile, "drift_av")[0] assert contextual_drift["audio_policy"]["lip_sync_options"] == lip_options color_drift, color_status = advanced.apply(combined, "color_drift_av") assert color_drift["transition_policy"][ "continuation_mode"] == "color_stable_drift_av" assert color_drift["audio_policy"] == combined["audio_policy"] assert "Color-Stable Drift AV" in color_status combined_plan = make_plan(combined=combined) assert "chain_policy" not in combined_plan["compatibility"] contextual_plan = make_plan(combined=contextual_profile) assert contextual_plan["compatibility"]["audio_policy"][ "lip_sync_options"] == lip_options legacy = chain.MiniMaxH3Legacy04PolicyAdapter() legacy_combined, legacy_status = legacy.build( "source_plus_timeline", "feathered_av", 39, 33) assert legacy.RETURN_NAMES == ("chain_policy", "status") assert legacy_combined["audio_policy"] == chain.migrate_legacy_audio_mode( "source_plus_timeline") assert legacy_combined["transition_policy"]["continuation_mode"] == ( "feathered_av") assert legacy_combined["audio_context_length"] == 33 assert "legacy 0.4 migration" in legacy_status assert legacy.INPUT_TYPES()["optional"]["chain_policy"][0] == ( chain.CHAIN_POLICY_TYPE) legacy_overlay, overlay_status = legacy.build( "generated_audio", "drift_control_av", 39, 39, chain_policy=combined) assert legacy_overlay["audio_policy"] == combined["audio_policy"] assert legacy_overlay["transition_policy"] == chain._contract_transition_policy( "drift_av") assert "incoming audio preserved" in overlay_status legacy_locked = legacy.build( "generated_audio", "masked_av", 39, 39, chain_policy=locked)[0] assert legacy_locked["audio_policy"] == locked["audio_policy"] legacy_plan = make_plan(combined=legacy_combined) assert legacy_plan["compatibility"]["audio_context_length"] == 33 assert legacy_plan["compatibility"]["continuation_mode"] == "feathered_av" plan_inputs = chain.MiniMaxH3ChainPlan.INPUT_TYPES() assert plan_inputs["optional"]["chain_policy"][0] == chain.CHAIN_POLICY_TYPE assert "audio_policy" not in plan_inputs["optional"] assert "transition_policy" not in plan_inputs["optional"] context_optional = chain.MiniMaxH3ChainContext.INPUT_TYPES()["optional"] assert list(context_optional) == [ "audio_vae", "model", "drift_sigmas", "lip_sync_voice"] assert context_optional["lip_sync_voice"][0] == "AUDIO" assert chain.CHAIN_NODE_CLASS_MAPPINGS[ "MiniMaxH3LipSyncOptions"] is chain.MiniMaxH3LipSyncOptions assert chain.CHAIN_NODE_CLASS_MAPPINGS[ "MiniMaxH3GenerationProfile"] is chain.MiniMaxH3GenerationProfile assert chain.CHAIN_NODE_CLASS_MAPPINGS[ "MiniMaxH3ChainPolicy"] is chain.MiniMaxH3ChainPolicy assert chain.CHAIN_NODE_CLASS_MAPPINGS[ "MiniMaxH3AdvancedPolicy"] is chain.MiniMaxH3AdvancedPolicy assert chain.CHAIN_NODE_DISPLAY_NAME_MAPPINGS[ "MiniMaxH3GenerationProfile"] == "MiniMax H3 Generation Profile" assert chain.CHAIN_NODE_DISPLAY_NAME_MAPPINGS[ "MiniMaxH3ChainPolicy"] == "MiniMax H3 Manual Chain Policy (Legacy)" assert chain.CHAIN_NODE_DISPLAY_NAME_MAPPINGS[ "MiniMaxH3AdvancedPolicy"] == "MiniMax H3 Advanced Policy Override" assert chain.CHAIN_NODE_DISPLAY_NAME_MAPPINGS[ "MiniMaxH3Legacy04PolicyAdapter"] == ( "MiniMax H3 Legacy 0.4 Policy Adapter") lora_plan = chain._normalize_plan( json.dumps({"shots": [ {"id": "base", "prompt": "Base scene.", "length": 73, "lora_route": "base"}, {"id": "hero", "prompt": "Hero scene.", "length": 73, "lora_route": "A"}, ]}), "lora-route-test", 64, 64, 22, "video", "head", "disabled", "generated_audio", 22, 3.0, 8, 7, 18, "model-stack", 0, "guide") assert "lora_route" not in lora_plan["shots"][0] assert lora_plan["shots"][1]["lora_route"] == "a" assert chain._effective_editor_plan(lora_plan)["shots"][1][ "lora_route"] == "a" assert chain._shot_lora_route(lora_plan["shots"][0]) == "base" try: chain._shot_lora_route({"lora_route": "hero"}) except ValueError as exc: assert "base" in str(exc) and "a" in str(exc) else: raise AssertionError("unknown scene LoRA route was accepted") try: chain._normalize_plan( json.dumps({"shots": [{ "id": "invalid", "prompt": "Invalid route.", "length": 73, "lora_route": "hero", }]}), "invalid-lora-route-test", 64, 64, 22, "video", "head", "disabled", "generated_audio", 22, 3.0, 8, 7, 18, "model-stack", 0, "guide") except ValueError as exc: assert "Shot 1" in str(exc) and "LoRA route" in str(exc) else: raise AssertionError("Plan accepted an unknown scene LoRA route") scheduler = chain.MiniMaxH3ChainLoRAScheduler() scheduler_inputs = scheduler.INPUT_TYPES() assert scheduler_inputs["required"]["base_model"][1]["lazy"] is True assert scheduler_inputs["optional"]["lora_a"][1]["lazy"] is True base_state = {"index": 1, "plan": lora_plan} hero_state = {"index": 2, "plan": lora_plan} base_model = object() hero_model = object() assert scheduler.check_lazy_status(base_state, None) == ["base_model"] assert scheduler.check_lazy_status( hero_state, None, lora_a=None) == ["lora_a"] assert scheduler.check_lazy_status( hero_state, None, lora_a=hero_model) == [] assert scheduler.select(base_state, base_model)[0] is base_model selected, selected_status = scheduler.select( hero_state, None, lora_a=hero_model) assert selected is hero_model assert "scene 2" in selected_status and "LoRA A" in selected_status try: scheduler.select(hero_state, None) except ValueError as exc: assert "lora_a input is not connected" in str(exc) else: raise AssertionError("unconnected selected LoRA route was accepted") assert chain.CHAIN_NODE_CLASS_MAPPINGS[ "MiniMaxH3ChainLoRAScheduler"] is chain.MiniMaxH3ChainLoRAScheduler assert chain.CHAIN_NODE_DISPLAY_NAME_MAPPINGS[ "MiniMaxH3ChainLoRAScheduler"] == "MiniMax H3 Scene LoRA Scheduler" print( "generation profiles, legacy manual policy, and lazy scene LoRA routing: " "clear one-wire Plan intent, canonical compatibility, and existing-loader " "MODEL selection pass")