LTX: older Image to Video pipelines get the text to video fallback on load; end-to-end test of the no-input chain

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Tw9fjYZhHqk196LykrWy3T

CHANGELOG.md

@@ -13,6 +13,7 @@
 
 ### Fixed
 
+- **An older LTX Image to Video pipeline now falls back to text to video** when no picture is fed in (its start slot is upgraded on load to bypass the anchor and use a plain stand-in), instead of refusing to start. The fallback chain is now also run end to end in a test.
 - **Cancelling during a Noise merging second pass now cancels** instead of being swallowed as a failed merge.
 - **Identity bank's lock** no longer counts probe passes (such as Noise merging's second answer) as steps of the run.
 - **Attention explorer ratings made on H3 no longer teach LTX** under the same Taste key: the two model families keep their ratings apart.

modules/models/ltx/pipeline.py

@@ -160,8 +160,27 @@ def _add_aspect(slots):
     return out
 
 
+def _add_text_fallback(slots):
+    """An Image to Video saved before text to video was the fallback: its start picture is required, so a run with nothing fed
+    in could not start. The start slot now says what to change when there is none (bypass the anchor, a plain picture of the
+    latent's size stands in), as the shipped preset does."""
+    ids = {x.get("id"): x for x in slots}
+    start, anchor, latent = ids.get("source_image"), ids.get("anchor"), ids.get("latent")
+    if not start or not anchor or anchor.get("node") != "LTXVImgToVideoInplace" or not latent or start.get("node") != "FunUILoadMedia":
+        return slots
+    roles = start.get("roles") or []
+    if any(r.get("text_fallback") for r in roles) or not any(r.get("input") == "media_id" for r in roles):
+        return slots
+    if "bypass" not in (anchor.get("inputs") or {}):
+        return slots
+    size = {"slot": "aspect", "width": "width", "height": "height"} if "aspect" in ids else {"slot": "latent", "width": "width", "height": "height"}
+    fallback = {"set": [{"slot": "anchor", "input": "bypass", "value": True}], "size": size}
+    new = [dict(r, text_fallback=fallback) if r.get("input") == "media_id" else r for r in roles]
+    return [dict(x, roles=new) if x is start else x for x in slots]
+
+
 def upgrade(slots, _outputs=None):
-    return _add_aspect(_upgrade_video(slots))
+    return _add_text_fallback(_add_aspect(_upgrade_video(slots)))
 
 
 def _upgrade_video(slots):

modules/models/ltx/tests/test_ltx.py

@@ -305,3 +305,18 @@ def test_unproven_enhancements_are_held_back_but_the_proven_ones_stay(comfyui):
     assert traits.missing_for(spec, traits_) == ["a proven run on this model (held back for now)"]
     other = type("S", (), {"id": "noise_merge", "requires": []})()
     assert traits.split([spec, other], traits_) == ([other], [spec])
+
+
+def test_an_older_image_to_video_gets_the_text_to_video_fallback(comfyui):
+    from modules.models.ltx import pipeline
+    slots = [
+        {"id": "latent", "node": "FunUIEmptyLatent", "inputs": {"width": 768, "height": 512}},
+        {"id": "source_image", "node": "FunUILoadMedia", "roles": [{"at": "assets.source_image", "input": "media_id", "video_frames": True}],
+         "inputs": {"media_id": ""}},
+        {"id": "anchor", "node": "LTXVImgToVideoInplace", "inputs": {"bypass": False}},
+    ]
+    fixed = pipeline.upgrade(slots)
+    fb = fixed[1]["roles"][0]["text_fallback"]
+    assert {"slot": "anchor", "input": "bypass", "value": True} in fb["set"] and fb["size"]["slot"] == "latent"
+    assert fixed[1]["roles"][0]["video_frames"] is True
+    assert pipeline.upgrade(fixed) == fixed

modules/models/ltx/tests/test_v2v_end_to_end.py

@@ -85,7 +85,7 @@ def _run(prompt, stubs):
         cls = comfy_nodes.NODE_CLASS_MAPPINGS[spec["class_type"]]
         kwargs = {k: value(v) for k, v in spec["inputs"].items()}
         if hasattr(cls, "GET_SCHEMA"):
-            out = cls.execute(**kwargs)
+            out = cls.EXECUTE_NORMALIZED(**kwargs)
             done[node_id] = tuple(out.result)
         else:
             done[node_id] = getattr(cls(), cls.FUNCTION)(**kwargs)
@@ -221,3 +221,31 @@ def test_image_to_video_takes_its_shape_from_the_aspect_source_picture(comfyui,
     assert shape[-2:] == (sized[1] // 32, sized[0] // 32)                              # the latent follows it
     sized, _ = latent_shape({"aspect_source": {"media_id": "", "blank_width": 256, "blank_height": 128}, "aspect": {"width": 256, "height": 128}, "latent": {"length": 9}})
     assert sized == (256, 128)                                                         # no picture: the project's own shape
+
+
+def test_image_to_video_with_nothing_fed_in_builds_its_latent_and_samples(comfyui, real_tiny_ltx, monkeypatch):
+    """What the app sends with no reference picture: stand-in pictures, the anchor bypassed. The whole chain up to the sampler runs."""
+    import asyncio
+    from pathlib import Path
+    import nodes as comfy_nodes
+    from core import graph
+    from modules.models.ltx.pipeline import presets
+
+    async def load():
+        await comfy_nodes.init_extra_nodes(init_custom_nodes=False)
+        await comfy_nodes.load_custom_node(str(Path(__file__).resolve().parents[4]), module_parent="custom_nodes")
+    asyncio.run(load())
+    slots = next(p["slots"] for p in presets() if p["id"] == "ltx23_image_to_video_references")
+    edits = {"aspect_source": {"media_id": "", "blank_width": 256, "blank_height": 128},
+             "aspect": {"width": 256, "height": 128}, "latent": {"length": 17},
+             "source_image": {"media_id": "", "blank_width": 256, "blank_height": 128},
+             "anchor": {"bypass": True}}
+    slots, problems = graph.override(slots, edits)
+    assert problems == []
+    prompt, _ = graph.build(slots)
+    cond = [[torch.randn(1, 5, 48), {}]]
+    run, done = _run(prompt, {"model": (real_tiny_ltx, ""), "vae": (StubVae(), ""), "audio_vae": (StubAudioVae(), ""),
+                              "clip": (None, ""), "positive": (cond, ""), "negative": (cond, "")})
+    run("join")
+    video, audio = done["join"][0]["samples"].unbind()
+    assert video.shape[-2:] == (4, 8)