nodes/exportAnything-ComfyUI-DiffusionGemmaPromptBuilder
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add MiniMax H3 reference-policy controls

Browse Source
This commit is contained in:
danrh
2026-09-04 05:33:03 -05:00
parent db71a81524
commit 431f3965db
13 changed files with 2967 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -4,6 +4,7 @@ __pycache__/
*.log
.agents/
.codex/
.codex_qa/
.venv/
venv/
output/
nodes.py
+1545 -132
View File
File diff suppressed because it is too large Load Diff
scripts/smoke_prompt_builder.py
+3 -1
View File
@@ -258,14 +258,16 @@ def test_minimax_h3_contract_and_splitter_compatibility():
"minimax_h3_prompt",
"candidate_ltx_prompt",
"candidate_negative_prompt",
"candidate_minimax_h3_prompt",
)
assert len(result) == 16, result
assert len(result) == 17, result
assert result[13] == h3_prompt, result[13]
assert "\n\noverall_soundscape:" in result[13], result[13]
assert "\n\nnon_diegetic_music:" in result[13], result[13]
assert result[0] == "", result[0]
assert result[1] == "", result[1]
assert result[2] == "", result[2]
assert result[16] == h3_prompt, result[16]
assert result[8] is True, _metadata(result)
assert result[12] is True, _metadata(result)
assert result[14] == "", result[14]
tests/test_advertising_contract_nodes.py
+1 -1
View File
@@ -307,7 +307,7 @@ class LegacyArtifactImmutabilityTests(unittest.TestCase):
EXPECTED = {
"audio_production_nodes.py": "3f0d1f5b3f5e002d6944a6eb914c6a46c0fe4679828f3546fd672a8b1a700e21",
"production_planning_nodes.py": "7ae98e808062aab16d8a470c5c8143c9881333354167b79a4d88a3110ab6e1e8",
"nodes.py": "fecd6ba7e28e1f286171e348511b24dc3a7537b296cf568cf8fca540e5b23ed0",
"nodes.py": "fb474fd2848bbda5df82b2e5d1568125bee04efa7faaf30dfdb41db2be023452",
"examples/15_minimax_h3_ref2va_music_video_v6.json": "594f6aef94531f87b74659e1e8004cccdff00716d0cfe263cee6bf2149545c83",
}
tests/test_director_structural_repair.py
+79
View File
@@ -107,6 +107,85 @@ class DirectorStructuralRepairTests(unittest.TestCase):
self.assertIn("removed_trailing_commas:1", repairs)
self.assertEqual(parsed["minimax_h3_prompt"], "unchanged")
def test_live_t2va_invalid_dialogue_closer_escape_is_recovered(self) -> None:
raw = r'''<|final_json|>
{
"minimax_h3_prompt": "integrated_multimodal_description: [Shot 1] 2D hand-drawn cel animation, preserve drawn linework, cel shading, stylized proportions, and design continuity across every cut, no live action, photorealism, or 3D CGI. A young girl with wide, determined eyes clutches a tattered teddy bear to her chest, surrounded by a circle of shadowy, menacing enemies in a dim alley. The girl looks up with a defiant smirk and says, (S1) <d lang=\"English\">maybe you haven't met my friend yet...\</d>. The camera performs a slow zoom-in on her face to capture her growing resolve.\n\n[Shot 2] At 00:03.500 the camera cuts to the enemies, a group of figures with distorted features leaning back and pointing, erupt in mocking laughter. The camera is a static low-angle shot to emphasize their perceived dominance.\n\n[Shot 3] At 00:06.000 the smash cuts to a wide shot of the girl in the center of the frame, holding the teddy bear high. She screams with intense energy, (S2) <d lang=\"English\">TEDDY BEAR ACTIVATE</</d> as she hurls the bear into the night sky. The camera tracks the bear upward with a rapid tilt movement.\n\n[Shot 4] At 00:08.200 the camera cuts to the bear mid-air, which rapidly morphs with shifting mechanical plates and glowing blue circuitry, revealing a massive teddy bear-battle cyborg. The cyborg descends rapidly and lands with a heavy, metallic thud on its feet directly behind the girl, sending up a cloud of dust. The final frame holds on the girl standing confidently in front of the towering cyborg in a wide shot composition.\n\noverall_soundscape: Distant echoing laughter, the whistle of the bear being thrown, mechanical whirring during the transformation, and a heavy resonant metallic impact as the cyborg hits the ground.\n\nnon_diegetic_music: High-tension orchestral swell that builds during the throw and a sudden, powerful brass hit when the cyborg lands."
}
<|end_final_json|>'''
brief = (
"2d anime of a young girl holding a teddy bear surrounded by enemies. "
"She states maybe you haven't met my friend yet. She screams TEDDY BEAR "
"ACTIVATE as the bear transforms into a teddy bear-battle cyborg. "
"No subtitles or words."
)
packet, parse_valid, warning = self.nodes.repair_or_salvage_prompt_packet(
raw,
self.nodes.GemmaContext(
user_prompt=brief,
source="none",
media_metadata={"source": "none", "minimax_h3_mode": "t2va"},
),
self.h3_target(),
)
self.assertTrue(parse_valid)
self.assertEqual(warning, "")
self.assertEqual(packet["ltx_prompt"], "")
self.assertEqual(packet["ideogram_prompt"], "")
self.assertEqual(packet["negative_prompt"], "")
self.assertEqual(packet["scene_segments"], [])
self.assertIn("maybe you haven't met my friend yet", packet["minimax_h3_prompt"])
self.assertNotIn(r"\</d>", packet["minimax_h3_prompt"])
self.assertNotIn("lang=", packet["minimax_h3_prompt"])
self.assertNotIn("</</d>", packet["minimax_h3_prompt"])
self.assertEqual(packet["minimax_h3_prompt"].count("<d>"), 2)
self.assertEqual(packet["minimax_h3_prompt"].count("</d>"), 2)
validation_reasons = self.nodes._minimax_h3_prompt_validation_reasons(
packet["minimax_h3_prompt"],
10.0,
"2d anime girl; no subtitles or words",
)
self.assertNotIn("minimax_h3_dialogue_tags_invalid", validation_reasons)
self.assertNotIn("minimax_h3_timeline_invalid", validation_reasons)
metadata = packet["metadata"]
self.assertFalse(metadata["model_output_json_parse_valid"])
self.assertTrue(metadata["deterministic_transport_repair"])
self.assertIn(
"removed_invalid_h3_dialogue_closer_json_escape:1",
metadata["deterministic_transport_repairs"],
)
self.assertEqual(
self.nodes._minimax_h3_prompt_validation_reasons(
packet["minimax_h3_prompt"],
10.0,
brief,
),
[],
)
def test_h3_dialogue_closer_escape_recovery_rejects_non_single_key_packets(self) -> None:
canonical = (
"integrated_multimodal_description: [Shot 1] A locked camera holds as (S1) "
r"<d>[English] hello\</d>. The final frame holds on the centered subject.\n\n"
r"overall_soundscape: N/A\n\nnon_diegetic_music: N/A"
)
for raw in (
'{"minimax_h3_prompt":"' + canonical + '","metadata":{}}',
'{"minimax_h3_prompt":"' + canonical + '","minimax_h3_prompt":"duplicate"}',
'{"minimax_h3_prompt":"' + canonical.replace("hello", 'hello "ambiguous"') + '"}',
):
with self.subTest(raw=raw[-80:]):
packet, parse_valid, warning = self.nodes.repair_or_salvage_prompt_packet(
raw,
self.context(),
self.h3_target(),
)
self.assertFalse(parse_valid)
self.assertTrue(warning)
self.assertFalse(
packet.get("metadata", {}).get("deterministic_transport_repair", False)
)
def test_duplicate_keys_are_never_deterministically_repaired(self) -> None:
raw = (
'{"minimax_h3_prompt":"first","minimax_h3_prompt":"second",'
tests/test_h3_prompt_isolation.py
+336
View File
@@ -0,0 +1,336 @@
from __future__ import annotations
import copy
import importlib.util
import json
import sys
import unittest
import uuid
from pathlib import Path
from unittest.mock import patch
import torch
ROOT = Path(__file__).resolve().parents[1]
COMFY_ROOT = ROOT.parents[1]
def load_nodes_module():
if str(COMFY_ROOT) not in sys.path:
sys.path.insert(0, str(COMFY_ROOT))
package_name = f"diffusiongemma_h3_prompt_isolation_{uuid.uuid4().hex}"
spec = importlib.util.spec_from_file_location(
package_name,
ROOT / "__init__.py",
submodule_search_locations=[str(ROOT)],
)
if spec is None or spec.loader is None:
raise RuntimeError("could not load DiffusionGemma Prompt Builder")
package = importlib.util.module_from_spec(spec)
sys.modules[package_name] = package
spec.loader.exec_module(package)
return sys.modules[f"{package_name}.nodes"]
class H3PromptIsolationTests(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.nodes = load_nodes_module()
def context(self):
return self.nodes.GemmaContext(
user_prompt="Create a concise ten-second scene.",
source="none",
media_metadata={"source": "none", "duration_seconds": 10.0},
)
def h3_target(self):
return self.nodes.TargetProfileConfig(
target_profile="minimax_h3",
minimax_h3_mode="t2va",
)
def test_bare_thought_only_output_is_reasoning_not_candidate_answer(self) -> None:
raw = (
"thought\n"
"The user wants a MiniMax H3 video prompt from an image.\n"
"- `ltx_prompt`: Empty.\n"
"- `ideogram_prompt`: An alternate still-image prompt.\n"
"- `minimax_h3_prompt`: Draft the requested timeline.\n"
"*Refining&#x20;**`minimax_h3_prompt`**&#x20;content:*\n"
"[Shot 1] A close-up holds on the subject.\n"
"`non_diegetic_music`: Somber strings transition to fast-"
)
reasoning, answer = self.nodes._split_reasoning_and_answer(raw)
self.assertEqual(answer, "")
self.assertIn("The user wants a MiniMax H3 video prompt", reasoning)
self.assertIn("minimax_h3_prompt", reasoning)
self.assertIn("Somber strings transition to fast-", reasoning)
def test_minimal_h3_json_is_expanded_to_the_host_packet(self) -> None:
h3_prompt = (
"integrated_multimodal_description: [Shot 1] A locked medium shot "
"holds on a performer beneath a single spotlight. The final frame "
"holds on the performer centered beneath the unchanged light.\n\n"
"overall_soundscape: Quiet room tone and soft fabric movement.\n\n"
"non_diegetic_music: N/A"
)
packet, parse_valid, warning = self.nodes.repair_or_salvage_prompt_packet(
json.dumps({"minimax_h3_prompt": h3_prompt}),
self.context(),
self.h3_target(),
deterministic_transport_repair=True,
)
self.assertTrue(parse_valid)
self.assertEqual(warning, "")
self.assertEqual(packet["minimax_h3_prompt"], h3_prompt)
self.assertEqual(packet["ltx_prompt"], "")
self.assertEqual(packet["ideogram_prompt"], "")
self.assertEqual(packet["negative_prompt"], "")
self.assertEqual(packet["scene_segments"], [])
self.assertEqual(
packet["metadata"]["host_structural_repairs"],
[
"completed_inactive_packet_field:ltx_prompt",
"completed_inactive_packet_field:ideogram_prompt",
"completed_inactive_packet_field:negative_prompt",
"completed_inactive_packet_field:scene_segments",
],
)
self.assertTrue(packet["metadata"]["model_output_json_parse_valid"])
def test_unambiguous_three_digit_cut_timestamp_is_canonicalized(self) -> None:
prompt = (
"integrated_multimodal_description: [Shot 1] A locked wide shot "
"establishes the subject. [Shot 2] At 04:500, the camera cuts to "
"a close-up and remains locked-off in a stable view. The final frame "
"holds on the centered subject.\n\n"
"overall_soundscape: Quiet room tone.\n\n"
"non_diegetic_music: N/A"
)
repaired, repairs = self.nodes._repair_minimax_h3_t2va_contract_transport(
prompt,
"t2va",
)
self.assertIn("[Shot 2] At 00:04.500, the camera cuts to", repaired)
self.assertNotIn("At 04:500", repaired)
self.assertIn("canonicalized_timed_cut_timestamps", repairs)
def test_h3_model_prompt_excludes_inactive_target_contracts_and_controls(self) -> None:
sentinel_values = (
"DO_NOT_LEAK_LTX_CONTROL",
"DO_NOT_LEAK_IDEOGRAM_CONTROL",
"DO_NOT_LEAK_LTX_METADATA",
"DO_NOT_LEAK_IDEOGRAM_METADATA",
)
model_prompt = self.nodes._build_model_prompt(
"A performer turns toward a spotlight.",
self.nodes.DEFAULT_MASTER_PROMPT,
"minimax_h3",
{
"source": "none",
"duration_seconds": 10.0,
"minimax_h3_mode": "t2va",
"ltx_generation_mode": sentinel_values[2],
"ideogram_render_style": sentinel_values[3],
},
ltx_style=sentinel_values[0],
ideogram_render_style=sentinel_values[1],
ideogram_exact_text="DO_NOT_LEAK_IDEOGRAM_EXACT_TEXT",
thinking_mode="off",
minimax_h3_mode="t2va",
)
self.assertIn('"minimax_h3_prompt":', model_prompt)
self.assertIn(
"The JSON object must contain minimax_h3_prompt, and no other top-level keys.",
model_prompt,
)
for forbidden in (
"ltx_prompt",
"ideogram_prompt",
"scene_segments",
'"ltx_',
'"ltx25_',
'"ideogram_',
*sentinel_values,
"DO_NOT_LEAK_IDEOGRAM_EXACT_TEXT",
):
with self.subTest(forbidden=forbidden):
self.assertNotIn(forbidden, model_prompt)
def test_h3_refinement_prompt_excludes_inactive_target_contracts(self) -> None:
refinement_prompt = self.nodes._build_minimax_h3_refinement_prompt(
"A silent two-shot orbital rescue scene.",
"candidate",
10.0,
"auto_scene_audio",
["minimax_h3_ref_sections_invalid"],
"ref2va",
"<Picture 1>: sole visual reference",
)
self.assertIn('"minimax_h3_prompt"', refinement_prompt)
for inactive_field in (
'"ltx_prompt"',
'"ideogram_prompt"',
'"negative_prompt"',
'"scene_segments"',
'"metadata"',
):
with self.subTest(inactive_field=inactive_field):
self.assertNotIn(inactive_field, refinement_prompt)
def runtime(self):
return self.nodes.RuntimeConfig(
model_path="unused-local-checkpoint",
backend="transformers_inprocess",
dtype="auto",
quantization="none",
local_files_only=True,
unload_policy="unload_after_run",
max_memory_gb=20.0,
status={"ready": True, "supports_pixels": True},
)
def image_context(self):
return self.nodes.GemmaContext(
user_prompt="Describe the reference faithfully.",
images=torch.zeros((1, 8, 8, 3), dtype=torch.float32),
source="image",
media_metadata={
"source": "image",
"reference_image_count": 1,
"image_role": "primary_image_source",
},
)
def grounded_ledger(self, registry: dict) -> dict:
asset = registry["assets"][0]
sample = asset["samples"][0]
return {
"schema": "dg-grounding-ledger/1",
"analysis_status": "grounded",
"observed_facts": [
{
"fact_id": "fact-1",
"claim": "A directly visible subject is present.",
"confidence": "high",
"categories": ["object"],
"evidence": [
{
"asset_id": asset["asset_id"],
"sample_ordinal": sample["sample_ordinal"],
"source_frame_index": sample["source_frame_index"],
"timecode_seconds": sample["timecode_seconds"],
}
],
"typed_claims": [],
}
],
"inferred_facts": [],
"creative_additions": [],
"uncertainties": [],
"grounding_failure_reasons": [],
}
def backend_result(self, ledger: dict, stage: str):
return self.nodes.BackendRunResult(
decoded_text=json.dumps(ledger),
transport={
"pixel_transport_confirmed": True,
"not_applicable": False,
"mismatch_reasons": [],
},
telemetry={
"schema_version": "dg-denoising-telemetry/1",
"forward_count": 2,
"snapshots": [{"canvas": 1, "step": 1}],
"stream_ended": True,
"final_output": {"text": [], "token_count": [1]},
"errors": [],
"refusal": {"detected": False},
},
effective_sampling={"profile": "checkpoint_defaults"},
timing={"total_seconds": 0.01},
stage=stage,
)
@staticmethod
def packet(prompt: str, ready: bool) -> dict:
return {
"ltx_prompt": "",
"ideogram_prompt": "",
"minimax_h3_prompt": prompt,
"negative_prompt": "",
"scene_segments": [],
"metadata": {
"target_profile": "minimax_h3",
"ready_for_generation": ready,
"blocked_reasons": [] if ready else ["json_parse_invalid"],
},
}
def test_audit_retry_forces_thinking_off_in_call_and_backend_options(self) -> None:
context = self.image_context()
registry = self.nodes._grounding_asset_registry(context)
ledger = self.grounded_ledger(registry)
invalid_packet = self.packet("salvaged invalid H3 output", False)
valid_packet = self.packet("strict JSON retry result", True)
observed_calls: list[tuple[str, str, str]] = []
def audit_legacy(*args, **kwargs):
options = kwargs["backend_options"]
options.call_budget.claim()
observed_calls.append((args[9], options.thinking_mode, options.stage))
kwargs["backend_results"].append(
self.backend_result(ledger, options.stage)
)
if options.stage == "audit_director":
return (
copy.deepcopy(invalid_packet),
"thought\nunfinished planning",
"unfinished planning",
"",
False,
"salvaged malformed output",
copy.deepcopy(invalid_packet["metadata"]),
)
packet = copy.deepcopy(valid_packet)
return packet, "compiler raw", "", "", True, "", packet["metadata"]
with patch.object(
self.nodes,
"_run_generation_packet_legacy",
side_effect=audit_legacy,
) as legacy, patch.object(self.nodes, "_maybe_unload"):
result = self.nodes._run_generation_packet(
self.runtime(),
context,
self.h3_target(),
thinking_mode="on",
grounding_guard_config={
"mode": "audit",
"retry_on_uncertain": True,
},
)
self.assertEqual(legacy.call_count, 2)
self.assertEqual(
observed_calls,
[
("on", "on", "audit_director"),
("off", "off", "audit_director_retry"),
],
)
self.assertEqual(result[0]["minimax_h3_prompt"], "strict JSON retry result")
if __name__ == "__main__":
unittest.main()
tests/test_ltx_creativity_and_splitter_diagnostics.py
+104 -6
View File
@@ -439,11 +439,15 @@ class LtxCreativityAndSplitterDiagnosticTests(unittest.TestCase):
"She takes one measured step while the locked camera remains steady."
)
candidate_negative = "extra people, duplicate subject, subtitles"
candidate_h3 = (
"integrated_multimodal_description: [Shot 1] A locked view holds.\n\n"
"overall_soundscape: N/A\n\nnon_diegetic_music: N/A"
)
grounding_reasons = ["visual_grounding_unverified"]
packet = {
"ltx_prompt": candidate_ltx,
"ideogram_prompt": "",
"minimax_h3_prompt": "",
"minimax_h3_prompt": candidate_h3,
"negative_prompt": candidate_negative,
"scene_segments": [],
"metadata": {
@@ -497,19 +501,113 @@ class LtxCreativityAndSplitterDiagnosticTests(unittest.TestCase):
),
)
self.assertEqual(
self.nodes.DiffusionGemmaJSONSplitter.RETURN_NAMES[14:16],
("candidate_ltx_prompt", "candidate_negative_prompt"),
self.nodes.DiffusionGemmaJSONSplitter.RETURN_NAMES[14:17],
(
"candidate_ltx_prompt",
"candidate_negative_prompt",
"candidate_minimax_h3_prompt",
),
)
self.assertEqual(
self.nodes.DiffusionGemmaJSONSplitter.RETURN_TYPES[14:16],
("STRING", "STRING"),
self.nodes.DiffusionGemmaJSONSplitter.RETURN_TYPES[14:17],
("STRING", "STRING", "STRING"),
)
self.assertEqual(len(split), 16)
self.assertEqual(len(split), 17)
self.assertEqual(split[0], "")
self.assertEqual(split[2], "")
self.assertFalse(split[12])
self.assertEqual(split[14], candidate_ltx)
self.assertEqual(split[15], candidate_negative)
self.assertEqual(split[16], "")
h3_candidate = "integrated_multimodal_description: [Shot 1] incomplete"
h3_packet = {
"ltx_prompt": "",
"ideogram_prompt": "",
"minimax_h3_prompt": h3_candidate,
"negative_prompt": "",
"scene_segments": [],
"metadata": {
"target_profile": "minimax_h3",
"ready_for_generation": False,
"blocked_reasons": ["minimax_h3_missing_overall_soundscape"],
},
}
h3_target = self.nodes.TargetProfileConfig(
target_profile="minimax_h3",
minimax_h3_mode="t2va",
)
h3_split = self.nodes.DiffusionGemmaJSONSplitter().split(
json.dumps(h3_packet),
context,
h3_target,
)
self.assertEqual(h3_split[13], "")
self.assertEqual(h3_split[16], h3_candidate)
def test_generation_gate_warn_policy_only_bypasses_h3_validation(self) -> None:
gate = self.nodes.DiffusionGemmaGenerationGate()
candidate = "A non-empty native MiniMax-H3 prompt candidate."
h3_metadata = json.dumps(
{
"target_profile": "minimax_h3",
"blocked_reasons": [
"minimax_h3_cut_transition_invalid",
"minimax_h3_dialogue_speaker_invalid",
],
}
)
with self.assertRaises(ValueError):
gate.gate(candidate, False, h3_metadata, "strict")
self.assertEqual(
gate.gate(candidate, False, h3_metadata, "warn_and_continue"),
(candidate,),
)
with self.assertRaisesRegex(ValueError, "json_parse_invalid"):
gate.gate(
candidate,
False,
json.dumps(
{
"blocked_reasons": [
"minimax_h3_cut_transition_invalid",
"json_parse_invalid",
]
}
),
"warn_and_continue",
)
with self.assertRaisesRegex(ValueError, "did not pass validation"):
gate.gate(
"",
False,
h3_metadata,
"warn_and_continue",
)
def test_generation_gate_warn_policy_never_bypasses_strict_grounding(self) -> None:
gate = self.nodes.DiffusionGemmaGenerationGate()
metadata = {
"blocked_reasons": ["minimax_h3_cut_transition_invalid"],
"grounding_guard": {
"schema": "dg-grounding-report/1",
"config": {"mode": "strict"},
"analysis_status": "uncertain",
"decision": "block",
"would_block": True,
"grounding_guard_would_block": True,
"blocked_reasons": ["visual_grounding_unverified"],
},
}
with self.assertRaisesRegex(ValueError, "Grounding Guard blocked generation"):
gate.gate(
"A non-empty candidate.",
False,
json.dumps(metadata),
"warn_and_continue",
)
if __name__ == "__main__":
tests/test_minimax_h3_dialogue.py
+219 -4
View File
@@ -260,8 +260,12 @@ class MiniMaxH3DialogueTests(unittest.TestCase):
minimax_h3_dialogue_line_count=2,
)
self.assertIn("<Subject 1> (S1)", model_prompt)
self.assertIn("repeat the speaking '<Subject N> (Sx)' pair", model_prompt)
self.assertIn("never use a bare (Sx) after multiple Subject tags", model_prompt)
self.assertIn("repeat the speaking '<Subject N> (S1)' pair", model_prompt)
self.assertIn("never output literal (Sx)", model_prompt)
self.assertIn(
"never output literal (Sx) or use a bare numbered cue after multiple Subject tags",
model_prompt,
)
refinement_prompt = self.nodes._build_minimax_h3_refinement_prompt(
"Create a two-image reference sequence with two spoken lines.",
@@ -276,9 +280,13 @@ class MiniMaxH3DialogueTests(unittest.TestCase):
minimax_h3_dialogue_line_count=2,
native_prompt_only=True,
)
self.assertIn("'<Subject N> (Sx)'", refinement_prompt)
self.assertIn("'<Subject N> (S1)'", refinement_prompt)
self.assertIn("never output literal (Sx)", refinement_prompt)
self.assertIn("repeat that pair for every line", refinement_prompt)
self.assertIn("never use a bare (Sx) after multiple Subject tags", refinement_prompt)
self.assertIn(
"never output literal (Sx) or use a bare numbered cue after multiple Subject tags",
refinement_prompt,
)
def test_t2va_required_off_and_auto_modes(self) -> None:
prompt = self.t2va_prompt()
@@ -418,6 +426,64 @@ class MiniMaxH3DialogueTests(unittest.TestCase):
self.assertEqual(repaired_again, canonical)
self.assertEqual(second_repairs, [])
def test_t2va_dialogue_transport_repairs_live_lang_attribute_and_duplicate_close(self) -> None:
canonical = self.one_shot_t2va_prompt()
live_candidate = canonical.replace(
"<d>[English]",
'<d lang="English">',
).replace("</d>", "</</d>")
normalized = self.nodes._normalize_minimax_h3_prompt(live_candidate, "t2va")
contract_repaired, contract_repairs = (
self.nodes._repair_minimax_h3_t2va_contract_transport(
normalized,
"t2va",
)
)
repaired, repairs = self.nodes._repair_minimax_h3_dialogue_transport(
contract_repaired,
"t2va",
)
self.assertEqual(contract_repairs, [])
self.assertEqual(repaired, canonical)
self.assertEqual(
repairs,
[
"canonicalized_dialogue_language_attribute",
"canonicalized_dialogue_duplicate_prefix_close",
],
)
self.assertEqual(repaired.count("(S1)"), canonical.count("(S1)"))
self.assertEqual(repaired.count("(S2)"), canonical.count("(S2)"))
self.assertEqual(
self.validation_reasons(repaired, mode="required", shot_count="1"),
[],
)
repaired_again, second_repairs = self.nodes._repair_minimax_h3_dialogue_transport(
repaired,
"t2va",
)
self.assertEqual(repaired_again, canonical)
self.assertEqual(second_repairs, [])
def test_dialogue_transport_rejects_conflicting_attribute_and_body_languages(self) -> None:
canonical = self.one_shot_t2va_prompt()
ambiguous = canonical.replace(
"<d>[English]",
'<d lang="English">[French]',
1,
)
repaired, repairs = self.nodes._repair_minimax_h3_dialogue_transport(
ambiguous,
"t2va",
)
self.assertEqual(repaired, ambiguous)
self.assertEqual(repairs, [])
def test_dialogue_transport_leaves_ambiguous_unbalanced_and_nested_markup_unchanged(self) -> None:
for h3_mode, canonical in (
("t2va", self.t2va_prompt()),
@@ -507,6 +573,155 @@ class MiniMaxH3DialogueTests(unittest.TestCase):
self.assertIn("minimax_h3_dialogue_tags_invalid", reasons)
self.assertNotIn("minimax_h3_dialogue_count_mismatch", reasons)
def test_t2va_contract_transport_repairs_the_attached_failure_without_rewriting_it(self) -> None:
brief = (
"2d anime of a young girl holding a teddy bear surrounded by enemies. She states: "
"\"maybe you haven't met my friend yet...\" The camera cuts to a her enemies - they are laughing "
"at the girl. Smash Cut to a wide shot of the same girl holding her teddy bear - she screams: "
"\"TEDDY BEAR ACTIVATE\" as she throws him up into the sky. Just then, the teddy bear morphs in "
"mid-air into a teddy bear-battle cyborg and lands heavy on it's metallic feet just behind the girl. "
"No subtitles or words."
)
candidate = (
"integrated_multimodal_description: [Shot 1] 2D hand-drawn cel animation, preserve drawn linework, "
"cel shading, stylized proportions, and design continuity across every cut, no live action, "
"photorealism, or 3D CGI. A young girl with determined eyes stands in a dark urban alleyway, "
"clutching a tattered teddy bear to her chest. Shadowy, menacing enemies surround her in a circle. "
"The girl looks up with a slight smirk and says (Sx: Girl) <d>English>maybe you haven't met my friend "
"yet...</d>. The camera performs a slow dramatic zoom-in on her face.\n"
"[Shot 2] At 00:03.500, the camera cuts to a medium shot of the enemies; they are jagged figures with "
"glowing eyes, laughing hysterically and pointing at the small girl. The camera uses a handheld-style "
"shake to emphasize their chaotic energy.\n"
"[Shot 3] At 00:06.000, the camera cuts to a wide shot of the girl in the center of the frame. She "
"raises the teddy bear high and screams (Sx: Girl) <d>English>TEDDY BEAR ACTIVATE</d> as she flurls the "
"toy into the night sky. The camera tilts upward rapidly to follow the bear's arc.\n"
"[Shot 4] At 00:07.500, the teddy bear morphs in mid-air, unfolding into mechanical plates and glowing "
"blue circuits, transforming into a massive metallic teddy bear-battle cyborg. The cyborg descends and "
"lands with a heavy metallic thud on its feet directly behind the girl, causing the ground to crack. "
"The final frame holds on a wide shot of the girl standing confidently in front of the towering cyborg "
"as the enemies recoil.\n\n"
"overall_soundscape: Wind whistling through the alley, the mocking laughter of the enemies, the girl's "
"defiant scream, mechanical whirring and clicking during transformation, and a thunderous metallic "
"impact as the cyborg hits the ground.\n\n"
"non_diegetic_music: A tense orchestral swell that builds in intensity, culminating in a heavy, "
"cinematic brass drop when the bear transforms and lands."
)
normalized = self.nodes._normalize_minimax_h3_prompt(candidate, "t2va")
repaired, repairs = self.nodes._repair_minimax_h3_t2va_contract_transport(
normalized,
"t2va",
)
repaired, dialogue_repairs = self.nodes._repair_minimax_h3_dialogue_transport(
repaired,
"t2va",
)
self.assertEqual(
set(repairs),
{
"canonicalized_timed_cut_openers",
"restored_dialogue_language_brackets",
"canonicalized_dialogue_speaker_ids",
},
)
self.assertEqual(dialogue_repairs, [])
self.assertEqual(repaired.count("(S1)"), 2)
self.assertNotIn("(Sx", repaired)
self.assertIn("<d>[English]maybe you haven't met my friend yet...</d>", repaired)
self.assertIn(
"[Shot 4] At 00:07.500, the shot cuts to: the teddy bear morphs",
repaired,
)
self.assertIn("teddy bear-battle cyborg", repaired)
self.assertFalse(self.nodes._minimax_h3_battle_intent_requested(brief))
self.assertEqual(
self.nodes._minimax_h3_prompt_validation_reasons(
repaired,
10.0,
brief,
),
[],
)
repaired_again, second_repairs = (
self.nodes._repair_minimax_h3_t2va_contract_transport(repaired, "t2va")
)
self.assertEqual(repaired_again, repaired)
self.assertEqual(second_repairs, [])
def test_t2va_contract_transport_leaves_ambiguous_bare_speakers_unchanged(self) -> None:
ambiguous = self.one_shot_t2va_prompt().replace("(S1)", "(Sx)").replace(
"(S2)",
"(Sx)",
)
repaired, repairs = self.nodes._repair_minimax_h3_t2va_contract_transport(
ambiguous,
"t2va",
)
self.assertEqual(repaired, ambiguous)
self.assertEqual(repairs, [])
def test_t2va_contract_transport_never_rewrites_cue_like_spoken_text(self) -> None:
prompt = self.one_shot_t2va_prompt().replace(
"We leave at dawn.",
"The literal token (Sx: Stage note) stays in my words.",
).replace(
"and Ivo (S2) replies",
"and Ivo replies",
)
repaired, _repairs = self.nodes._repair_minimax_h3_t2va_contract_transport(
prompt,
"t2va",
)
self.assertIn("<d>[English] The literal token (Sx: Stage note) stays", repaired)
def test_timed_smash_cut_alias_is_canonical_and_idempotent(self) -> None:
timeline = (
"[Shot 1] A locked wide shot holds. "
"[Shot 2] At 00:04.000, SMASH CUT TO: A close-up reveals the switch."
)
repaired = self.nodes._repair_minimax_h3_timed_cut_openers(timeline)
self.assertIn(
"[Shot 2] At 00:04.000, the camera hard cuts to A close-up",
repaired,
)
self.assertEqual(
self.nodes._repair_minimax_h3_timed_cut_openers(repaired),
repaired,
)
def test_live_t2va_smash_cut_without_comma_is_canonical_and_idempotent(self) -> None:
timeline = (
"[Shot 1] A slow zoom holds on the girl. "
"[Shot 2] At 00:06.000 the smash cuts to a wide shot as she throws the bear."
)
repaired = self.nodes._repair_minimax_h3_timed_cut_openers(timeline)
self.assertIn(
"[Shot 2] At 00:06.000, the camera hard cuts to a wide shot",
repaired,
)
self.assertEqual(
self.nodes._repair_minimax_h3_timed_cut_openers(repaired),
repaired,
)
def test_battle_intent_ignores_character_labels_but_keeps_real_combat(self) -> None:
self.assertFalse(
self.nodes._minimax_h3_battle_intent_requested(
"A teddy bear-battle cyborg lands behind the girl."
)
)
self.assertFalse(
self.nodes._minimax_h3_battle_intent_requested(
"A portrait introduces the battle cyborg design."
)
)
self.assertTrue(
self.nodes._minimax_h3_battle_intent_requested(
"Two knights battle in the courtyard."
)
)
@staticmethod
def dialogue_patch_payload(records: list[dict], **extra: object) -> str:
return json.dumps(
tests/test_minimax_h3_ref2va_deterministic_contract_repair.py
+190
View File
@@ -420,6 +420,196 @@ class MiniMaxH3Ref2VADeterministicContractRepairTests(unittest.TestCase):
self.assertIn("<Subject 1> (derived from <Picture 1>)", timeline)
self.assertNotIn("minimax_h3_ref_subject_usage_invalid", self.reasons(repaired))
def test_live_dropped_n_music_alias_with_empty_canonical_footer_is_repaired(self) -> None:
music_heading = "non_diegetic_music:\n"
candidate = self.valid_prompt().replace(
music_heading,
"on_diegetic_music:\n",
1,
) + "\n\nnon_diegetic_music:"
self.assertIn("minimax_h3_extra_top_level_field", self.reasons(candidate))
self.assertIn("minimax_h3_ref_missing_non_diegetic_music", self.reasons(candidate))
repaired = self.nodes._repair_minimax_h3_ref2va_structure(
candidate,
self.user_prompt(),
self.manifest(),
15.0,
)
self.assertEqual(repaired.count("non_diegetic_music:"), 1)
self.assertNotRegex(repaired, r"(?m)^on_diegetic_music:")
self.assertIn("The exact locked fast-paced rap track", repaired)
self.assertEqual(self.reasons(repaired), [])
def test_duplicate_empty_music_footer_preserves_complete_music_section(self) -> None:
candidate = self.valid_prompt() + "\n\nnon_diegetic_music:"
repaired = self.nodes._repair_minimax_h3_ref2va_structure(
candidate,
self.user_prompt(),
self.manifest(),
15.0,
)
self.assertEqual(repaired.count("non_diegetic_music:"), 1)
self.assertIn("The exact locked fast-paced rap track", repaired)
self.assertEqual(self.reasons(repaired), [])
def test_live_cosmic_environment_subject_gets_a_timeline_binding(self) -> None:
user_prompt = (
"No dialogue. Superman holds the lifeless Lois Lane in space, overcome with guilt and grief before his "
"sadness turns to anger and he rushes back to Earth as a streak of energy."
)
manifest = (
"<Picture 1>: [dg:identity,appearance,color,environment,lighting,composition] sole visual reference for "
"Superman, Lois Lane, their visible appearance, colors, spatial relationship, space environment, lighting, "
"and composition; preserve visible identities and scene attributes throughout."
)
candidate = (
"subject_definitions:\n"
"<Subject 1>: (Picture 1): Superman in his classic blue suit with a red cape and the yellow 'S' shield, "
"expressing intense grief and burgeoning rage.\n"
"<Subject 2>: (Picture 1): Lois Lane, with long wavy hair, wearing a white maternity dress, appearing limp "
"and lifeless in space.\n"
"<Subject 3>: The void of space above the curved horizon of Earth, with clouds and distant stars visible.\n"
"<Picture 1>: sole visual reference for all reusable visible content and scene attributes; picture count "
"does not determine semantic Subject count.\n\n"
"summary:\n"
"[reference generation] The scene depicts Superman holding the lifeless Lois Lane in the orbit of Earth, "
"overcome by guilt and grief before transitioning into a furious, high-speed descent back to Earth.\n\n"
"retention_analysis:\n"
"<Subject 1>: fully_preserved - Superman's appearance and emotional state are maintained from Picture 1.\n"
"<Subject 2>: fully_preserved - Lois Lane's appearance and white dress are maintained from Picture 1.\n"
"<Subject 3>: fully_preserved - The space environment and Earth horizon are consistent with Picture 1.\n"
"<Picture 1>: attribute_transfer - sole visual reference for all reusable visible content and scene "
"attributes; picture count does not determine semantic Subject count.\n\n"
"detailed_description:\n"
"The visual style is cinematic with high-contrast lighting, utilizing deep shadows of space and the "
"brilliant glow from the Earth's atmosphere. [Shot 1] In the silent void of space, <Subject 1> holds the "
"limp <Subject 2> in his arms; <Subject 1>'s face is contorted in silent anguish as he looks down at "
"<Subject 2>, whose white dress drifts slightly in zero-gravity. The camera performs a slow, tragic orbit "
"around the pair, capturing their isolation against the vast, curved Earth below. [Shot 2] At 00:05.000, "
"the shot cuts to: <Subject 1>'s expression shifts from sobbing to a fierce snarl; he releases <Subject 2> "
"and ignites toward the camera. The camera executes a rapid tracking shot following him as he transforms "
"into a blurred streak of blue and red energy, creating heavy motion blur as he breaks into the upper "
"atmosphere.\n\n"
"overall_soundscape:\n"
"The low, rhythmic hum of space vacuum, followed by the sudden sonic boom and rushing roar of air friction "
"as Superman enters the atmosphere.\n\n"
"non_diegetic_music:\n"
"A dramatic, orchestral score starting with somber, mourn cello notes building into a fast, aggressive "
"percussion and brass crescendo as he descends."
)
def live_reasons(prompt: str) -> list[str]:
return self.nodes._minimax_h3_ref2va_validation_reasons(
prompt,
10.0,
user_prompt,
"auto_scene_audio",
20_000,
manifest,
"auto",
"auto",
2,
"",
0,
)
self.assertEqual(
live_reasons(candidate),
["minimax_h3_ref_subject_usage_invalid"],
)
repaired = self.nodes._repair_minimax_h3_ref2va_structure(
candidate,
user_prompt,
manifest,
10.0,
)
repaired_sections = self.nodes._minimax_h3_ref_sections(repaired)
self.assertIsNotNone(repaired_sections)
self.assertIn(
"[Shot 1] The setting is <Subject 3> as defined above. In the silent void of space",
repaired_sections["detailed_description"],
)
self.assertEqual(live_reasons(repaired), [])
self.assertEqual(
self.nodes._repair_minimax_h3_ref2va_structure(
repaired,
user_prompt,
manifest,
10.0,
),
repaired,
)
def test_multiple_implicit_environment_subjects_remain_fail_closed(self) -> None:
sections = self.nodes._minimax_h3_ref_sections(self.valid_prompt())
self.assertIsNotNone(sections)
sections = copy.deepcopy(sections)
sections["subject_definitions"] = sections["subject_definitions"].replace(
"<Audio 1>:",
"<Subject 2>: The minimalist tiled room environment with deep shadows and hard side lighting.\n"
"<Subject 3>: A separate exterior city location under blue evening light and drifting fog.\n"
"<Audio 1>:",
1,
)
sections["retention_analysis"] = sections["retention_analysis"].replace(
"<Audio 1>:",
"<Subject 2>: fully_preserved - The tiled room and hard side lighting remain stable.\n"
"<Subject 3>: fully_preserved - The city exterior and blue evening fog remain stable.\n"
"<Audio 1>:",
1,
)
candidate = self.nodes._rebuild_minimax_h3_ref_prompt(sections)
repaired = self.nodes._repair_minimax_h3_ref2va_structure(
candidate,
self.user_prompt(),
self.manifest(),
15.0,
)
repaired_sections = self.nodes._minimax_h3_ref_sections(repaired)
self.assertIsNotNone(repaired_sections)
self.assertNotIn("<Subject 2>", repaired_sections["detailed_description"])
self.assertNotIn("<Subject 3>", repaired_sections["detailed_description"])
self.assertIn("minimax_h3_ref_subject_usage_invalid", self.reasons(repaired))
def test_implicit_style_subject_is_not_recast_as_the_setting(self) -> None:
sections = self.nodes._minimax_h3_ref_sections(self.valid_prompt())
self.assertIsNotNone(sections)
sections = copy.deepcopy(sections)
sections["subject_definitions"] = sections["subject_definitions"].replace(
"<Audio 1>:",
"<Subject 2>: A visual style palette with sharp film grain and hard monochrome lighting.\n"
"<Audio 1>:",
1,
)
sections["retention_analysis"] = sections["retention_analysis"].replace(
"<Audio 1>:",
"<Subject 2>: fully_preserved - The visual style, palette, film grain, and lighting remain stable.\n"
"<Audio 1>:",
1,
)
candidate = self.nodes._rebuild_minimax_h3_ref_prompt(sections)
repaired = self.nodes._repair_minimax_h3_ref2va_structure(
candidate,
self.user_prompt(),
self.manifest(),
15.0,
)
repaired_sections = self.nodes._minimax_h3_ref_sections(repaired)
self.assertIsNotNone(repaired_sections)
self.assertNotIn("<Subject 2>", repaired_sections["detailed_description"])
self.assertIn("minimax_h3_ref_subject_usage_invalid", self.reasons(repaired))
def test_salvage_keeps_locked_audio_as_fully_copy(self) -> None:
sections = self.nodes._minimax_h3_ref_sections(self.valid_prompt())
self.assertIsNotNone(sections)
tests/test_minimax_h3_reference_manifest_presets.py
+84
View File
@@ -143,6 +143,8 @@ class MiniMaxH3ReferenceManifestPresetTests(unittest.TestCase):
"custom",
"1 image - all visual attributes",
"2 images - subject + environment/style",
"2 images - main subject + same-subject contact sheet",
"2 images - main subject + multi-entity contact sheet",
],
)
self.assertEqual(config["default"], "custom")
@@ -332,6 +334,88 @@ class MiniMaxH3ReferenceManifestPresetTests(unittest.TestCase):
self.ref2va_validation_reasons(prompt),
)
def test_multi_entity_contact_sheet_enforces_subject_source_bindings(self) -> None:
manifest = self.nodes._resolve_minimax_h3_reference_manifest(
"",
"2 images - main subject + multi-entity contact sheet",
)
invalid = self.valid_ref2va_prompt(subject_count=2, picture_count=2)
invalid_reasons = self.nodes._minimax_h3_ref2va_validation_reasons(
invalid,
12.0,
"Create exactly two shots in a quiet train compartment.",
"auto_scene_audio",
8000,
manifest,
"2",
"required",
2,
"Mara says We leave at dawn, then says Then we are ready.",
minimax_h3_expected_subject_count=2,
)
self.assertIn(
"minimax_h3_ref_contact_sheet_subject_source_invalid",
invalid_reasons,
)
valid = invalid.replace(
"derive from <Picture 1> and <Picture 2>",
"derive from <Picture 1>",
).replace(
"derived from <Picture 1> and <Picture 2>",
"derived from <Picture 1>",
).replace(
"A recurring blue geometric compartment fixture with stable brass trim derived from <Picture 1>.",
"A recurring blue geometric compartment fixture with stable brass trim derived from <Picture 2>.",
)
valid_reasons = self.nodes._minimax_h3_ref2va_validation_reasons(
valid,
12.0,
"Create exactly two shots in a quiet train compartment.",
"auto_scene_audio",
8000,
manifest,
"2",
"required",
2,
"Mara says We leave at dawn, then says Then we are ready.",
minimax_h3_expected_subject_count=2,
)
self.assertEqual(valid_reasons, [])
def test_contact_sheet_layout_language_is_rejected_from_timeline(self) -> None:
manifest = self.nodes._resolve_minimax_h3_reference_manifest(
"",
"2 images - main subject + multi-entity contact sheet",
)
prompt = self.valid_ref2va_prompt(subject_count=2, picture_count=2).replace(
"derive from <Picture 1> and <Picture 2>",
"derive from <Picture 1>",
).replace(
"derived from <Picture 1> and <Picture 2>",
"derived from <Picture 1>",
).replace(
"A recurring blue geometric compartment fixture with stable brass trim derived from <Picture 1>.",
"A recurring blue geometric compartment fixture with stable brass trim derived from <Picture 2>.",
).replace(
"Realistic live-action practical film photography",
"A contact-sheet grid remains visible behind them. Realistic live-action practical film photography",
)
reasons = self.nodes._minimax_h3_ref2va_validation_reasons(
prompt,
12.0,
"Create exactly two shots in a quiet train compartment.",
"auto_scene_audio",
8000,
manifest,
"2",
"required",
2,
"Mara says We leave at dawn, then says Then we are ready.",
minimax_h3_expected_subject_count=2,
)
self.assertIn("minimax_h3_ref_contact_sheet_layout_transfer_invalid", reasons)
if __name__ == "__main__":
unittest.main()
tests/test_minimax_h3_reference_policy.py
+402
View File
@@ -0,0 +1,402 @@
from __future__ import annotations
import importlib.util
import json
import sys
import unittest
import uuid
from pathlib import Path
from unittest.mock import patch
import torch
ROOT = Path(__file__).resolve().parents[1]
COMFY_ROOT = ROOT.parents[1]
def load_nodes_module():
if str(COMFY_ROOT) not in sys.path:
sys.path.insert(0, str(COMFY_ROOT))
package_name = f"diffusiongemma_h3_reference_policy_{uuid.uuid4().hex}"
spec = importlib.util.spec_from_file_location(
package_name,
ROOT / "__init__.py",
submodule_search_locations=[str(ROOT)],
)
if spec is None or spec.loader is None:
raise RuntimeError("could not load DiffusionGemma Prompt Builder")
package = importlib.util.module_from_spec(spec)
sys.modules[package_name] = package
spec.loader.exec_module(package)
return sys.modules[f"{package_name}.nodes"]
class MiniMaxH3ReferencePolicyTests(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.nodes = load_nodes_module()
def target(self, mode: str = "ref2va"):
return self.nodes.TargetProfileConfig(
target_profile="minimax_h3",
minimax_h3_mode=mode,
)
def test_policy_node_is_registered_and_director_socket_is_appended(self) -> None:
self.assertIn("DiffusionGemmaH3ReferencePolicy", self.nodes.NODE_CLASS_MAPPINGS)
node = self.nodes.DiffusionGemmaH3ReferencePolicy
self.assertEqual(
node.RETURN_TYPES,
(self.nodes.H3_REFERENCE_POLICY_TYPE, "STRING", "STRING"),
)
required = node.INPUT_TYPES()["required"]
self.assertEqual(
list(required),
["layout", "expected_subject_count", "custom_manifest"],
)
self.assertEqual(required["layout"][1]["default"], "Auto (recommended)")
director_optional = self.nodes.DiffusionGemmaCoTGenerator.INPUT_TYPES()["optional"]
self.assertEqual(list(director_optional)[-1], "h3_reference_policy")
self.assertEqual(
director_optional["h3_reference_policy"][0],
self.nodes.H3_REFERENCE_POLICY_TYPE,
)
def test_generic_context_one_image_autopopulates_manifest(self) -> None:
image = torch.zeros((1, 32, 48, 3), dtype=torch.float32)
context, _context_json, _preview = self.nodes.DiffusionGemmaContextHub().build(
"The referenced subject crosses a moonlit room.",
image=image,
)
self.assertNotIn("minimax_h3_reference_manifest", context.media_metadata)
prepared = self.nodes._context_with_minimax_h3_reference_policy(
context,
self.target(),
)
manifest = prepared.media_metadata["minimax_h3_reference_manifest"]
self.assertEqual(self.nodes._minimax_h3_reference_tags(manifest), ["<Picture 1>"])
self.assertEqual(
prepared.media_metadata["minimax_h3_reference_manifest_source"],
"director_auto",
)
self.assertEqual(
prepared.media_metadata["minimax_h3_reference_manifest_reasons"],
[],
)
self.assertNotIn("minimax_h3_reference_manifest", context.media_metadata)
def test_h3_context_treats_blank_inline_manifest_as_pending_auto(self) -> None:
image = torch.zeros((1, 32, 48, 3), dtype=torch.float32)
context, context_json, preview = self.nodes.DiffusionGemmaH3ReferenceContext().build(
"Animate the attached reference.",
"",
reference_images=image,
reference_manifest_preset="custom",
)
metadata = json.loads(context_json)["media"]
self.assertEqual(
context.media_metadata["minimax_h3_reference_manifest_source"],
"pending_auto",
)
self.assertEqual(metadata["minimax_h3_reference_manifest_reasons"], [])
self.assertNotIn("needs correction", "\n".join(context.warnings))
self.assertNotIn("manifest declares 0", "\n".join(context.warnings))
self.assertIn("H3 reference roles: Auto", preview)
def test_context_hub_records_the_real_image_batch_count(self) -> None:
images = torch.zeros((3, 32, 48, 3), dtype=torch.float32)
context, context_json, _preview = self.nodes.DiffusionGemmaContextHub().build(
"Use these three ordered references.",
image=images,
)
self.assertEqual(context.media_metadata["reference_image_count"], 3)
self.assertEqual(json.loads(context_json)["media"]["reference_image_count"], 3)
prepared = self.nodes._context_with_minimax_h3_reference_policy(
context,
self.target(),
)
self.assertEqual(
self.nodes._minimax_h3_reference_tags(
prepared.media_metadata["minimax_h3_reference_manifest"]
),
["<Picture 1>", "<Picture 2>", "<Picture 3>"],
)
def test_connected_contact_sheet_policy_overrides_stale_inline_manifest(self) -> None:
images = torch.zeros((2, 32, 48, 3), dtype=torch.float32)
stale_manifest = self.nodes._MINIMAX_H3_ONE_PICTURE_MANIFEST
context = self.nodes.GemmaContext(
user_prompt="The hero uses the two requested tools from the contact sheet.",
images=images,
source="image",
media_metadata={
"source": "image",
"reference_image_count": 2,
"reference_image_backend_attached": True,
"minimax_h3_reference_manifest": stale_manifest,
},
)
policy, _json, _preview = self.nodes.DiffusionGemmaH3ReferencePolicy().build(
"Primary subject + supporting subjects/objects contact sheet",
3,
"",
)
prepared = self.nodes._context_with_minimax_h3_reference_policy(
context,
self.target(),
policy,
)
metadata = prepared.media_metadata
self.assertEqual(metadata["minimax_h3_reference_manifest_source"], "policy_node")
self.assertEqual(metadata["minimax_h3_expected_subject_count"], 3)
self.assertIn("one multi-entity contact sheet", metadata["minimax_h3_reference_manifest"])
self.assertEqual(
self.nodes._minimax_h3_reference_tags(metadata["minimax_h3_reference_manifest"]),
["<Picture 1>", "<Picture 2>"],
)
self.assertNotEqual(metadata["minimax_h3_reference_manifest"], stale_manifest)
def test_contact_sheet_policy_requires_two_picture_assets(self) -> None:
policy = self.nodes._make_h3_reference_policy_config(
"Primary subject + supporting subjects/objects contact sheet"
)
with self.assertRaisesRegex(ValueError, "requires exactly 2 attached Picture"):
self.nodes._minimax_h3_reference_policy_manifest(policy, 1)
with self.assertRaisesRegex(ValueError, "contact sheet is one Picture asset"):
self.nodes._minimax_h3_reference_policy_manifest(policy, 3)
def test_custom_manifest_remains_explicit_and_invalid_instead_of_falling_back(self) -> None:
image = torch.zeros((1, 32, 48, 3), dtype=torch.float32)
context = self.nodes.GemmaContext(
user_prompt="Test custom roles.",
images=image,
source="image",
media_metadata={"source": "image"},
)
policy = self.nodes._make_h3_reference_policy_config(
"Custom manifest",
"this is intentionally not a tagged manifest",
)
prepared = self.nodes._context_with_minimax_h3_reference_policy(
context,
self.target(),
policy,
)
self.assertEqual(
prepared.media_metadata["minimax_h3_reference_manifest_source"],
"policy_node",
)
self.assertIn(
"minimax_h3_ref_manifest_missing",
prepared.media_metadata["minimax_h3_reference_manifest_reasons"],
)
def test_missing_visual_reference_fails_with_connection_guidance(self) -> None:
context = self.nodes.GemmaContext(user_prompt="Animate my reference.")
with self.assertRaisesRegex(ValueError, "no visual reference in DG_CONTEXT"):
self.nodes._context_with_minimax_h3_reference_policy(
context,
self.target(),
)
def test_non_ref2va_context_is_unchanged(self) -> None:
context = self.nodes.GemmaContext(user_prompt="Text-only H3 scene.")
self.assertIs(
self.nodes._context_with_minimax_h3_reference_policy(
context,
self.target("t2va"),
),
context,
)
def test_director_resolves_connected_policy_before_generation(self) -> None:
images = torch.zeros((2, 32, 48, 3), dtype=torch.float32)
context = self.nodes.GemmaContext(
user_prompt="The hero selects one requested object from the contact sheet.",
images=images,
source="image",
media_metadata={"source": "image", "reference_image_count": 2},
)
policy = self.nodes._make_h3_reference_policy_config(
"Primary subject + supporting subjects/objects contact sheet",
expected_subject_count=2,
)
runtime = self.nodes.RuntimeConfig(
model_path="unused",
backend="template",
dtype="auto",
quantization="none",
local_files_only=True,
unload_policy="unload_after_run",
max_memory_gb=0.0,
)
captured = {}
def fake_run(_runtime, prepared_context, *_args, **_kwargs):
captured["context"] = prepared_context
metadata = {
"ready_for_generation": True,
"grounding_guard": {
"analysis_status": "not_run",
"decision": "disabled",
},
}
packet = {
"ltx_prompt": "",
"ideogram_prompt": "",
"minimax_h3_prompt": "valid placeholder",
"negative_prompt": "",
"scene_segments": [],
"metadata": metadata,
}
return packet, "raw", "", "", True, "", metadata
with patch.object(self.nodes, "_run_generation_packet", side_effect=fake_run):
self.nodes.DiffusionGemmaCoTGenerator().generate(
runtime,
context,
self.target(),
temperature=0.45,
creativity_mode="faithful",
creative_strength=0.0,
thinking_mode="off",
director_cache_mode="off",
grounding_guard_mode="off",
h3_reference_policy=policy,
)
prepared = captured["context"]
self.assertEqual(
prepared.media_metadata["minimax_h3_reference_manifest_source"],
"policy_node",
)
self.assertIn(
"one multi-entity contact sheet",
prepared.media_metadata["minimax_h3_reference_manifest"],
)
def test_splitter_restores_director_resolved_policy_over_blank_context(self) -> None:
image = torch.zeros((1, 32, 48, 3), dtype=torch.float32)
context = self.nodes.GemmaContext(
user_prompt="Hold on the referenced traveler in one quiet shot.",
images=image,
source="image",
media_metadata={
"source": "image",
"reference_image_count": 1,
"reference_image_backend_attached": True,
"minimax_h3_reference_manifest": "",
"warnings": [
"H3 reference manifest needs correction: minimax_h3_ref_manifest_missing",
"DiffusionGemma received 1 reference picture(s) for analysis, while the manifest declares 0; the native H3 connections remain authoritative.",
"Media pixels will be passed to the active backend.",
],
},
)
policy = self.nodes._make_h3_reference_policy_config("Auto (recommended)")
prepared = self.nodes._context_with_minimax_h3_reference_policy(
context,
self.target(),
policy,
)
packet = {
"ltx_prompt": "",
"ideogram_prompt": "",
"minimax_h3_prompt": "valid diagnostic placeholder",
"negative_prompt": "",
"scene_segments": [],
"metadata": {"media": prepared.media_metadata},
}
split = self.nodes.DiffusionGemmaJSONSplitter().split(
json.dumps(packet),
context,
self.target(),
)
metadata = json.loads(split[4])
self.assertEqual(
metadata["media"]["minimax_h3_reference_manifest_source"],
"policy_node",
)
self.assertEqual(
metadata["media"]["minimax_h3_reference_tags"],
["<Picture 1>"],
)
self.assertNotIn(
"minimax_h3_ref_manifest_missing",
metadata["blocked_reasons"],
)
self.assertNotIn(
"minimax_h3_ref_undefined_tag",
metadata["blocked_reasons"],
)
self.assertEqual(
metadata["media"]["warnings"],
["Media pixels will be passed to the active backend."],
)
def test_policy_layout_changes_director_cache_key(self) -> None:
images = torch.zeros((2, 32, 48, 3), dtype=torch.float32)
context = self.nodes.GemmaContext(
user_prompt="Animate the ordered references.",
images=images,
source="image",
media_metadata={"source": "image", "reference_image_count": 2},
)
runtime = self.nodes.RuntimeConfig(
model_path="unused",
backend="template",
dtype="auto",
quantization="none",
local_files_only=True,
unload_policy="unload_after_run",
max_memory_gb=0.0,
)
environment_context = self.nodes._context_with_minimax_h3_reference_policy(
context,
self.target(),
self.nodes._make_h3_reference_policy_config(
"Primary subject + environment/style"
),
)
contact_context = self.nodes._context_with_minimax_h3_reference_policy(
context,
self.target(),
self.nodes._make_h3_reference_policy_config(
"Primary subject + supporting subjects/objects contact sheet"
),
)
common = {
"temperature": 0.45,
"creativity_mode": "faithful",
"creative_strength": 0.0,
"thinking_mode": "off",
"max_new_tokens": 1024,
}
environment_key = self.nodes._director_cache_key(
runtime,
environment_context,
self.target(),
{"mode": "off"},
**common,
)
contact_key = self.nodes._director_cache_key(
runtime,
contact_context,
self.target(),
{"mode": "off"},
**common,
)
self.assertNotEqual(environment_key, contact_key)
if __name__ == "__main__":
unittest.main()
tests/test_minimax_h3_shot_count.py
+2 -1
View File
@@ -312,7 +312,8 @@ class MiniMaxH3ShotCountTests(unittest.TestCase):
self.assertIn("immediately after its [Shot N] marker", model_prompt)
self.assertIn("An incoming cut, shot size, angle, subject framing", model_prompt)
self.assertIn("never omit it to save words in a dense multi-shot timeline", model_prompt)
self.assertIn("MiniMax H3 is not governed by the LTX Stable / base-model Camera Capability", model_prompt)
self.assertNotIn("LTX Stable / base-model Camera Capability", model_prompt)
self.assertIn("Honor explicit user and reference camera choreography exactly", model_prompt)
self.assertIn("orbit or arc, roll or rotation, spin or swirl", model_prompt)
self.assertIn("sweeping or whip movement", model_prompt)
self.assertIn("pronounced or layered parallax", model_prompt)
tests/test_splatstage_planner.py
+1
View File
@@ -34,6 +34,7 @@ EXPERIMENTAL_NODE_IDS = {
"DiffusionGemmaSplatStageEfficientPlanner",
}
OPTIONAL_NODE_IDS = {
"DiffusionGemmaH3ReferencePolicy",
"DiffusionGemmaReferencePrep",
"DiffusionGemmaH3ReferencePairPrep",
}