make a per-model setting reachable by the name it was set on

The sequential text-encoder offload reported itself "not enabled" on a card
that plainly had `sequential_text_encoder = True`. Four defects, stacked so
each one hid the next; the first is much wider than the feature that exposed
it.

_resolve_config_key could not resolve ANY non-text model by alias. Typed
entries are keyed "<kind>:<name>", so the exact-id and alias-map lookups both
miss, and the basename fallback misses too: "video:LTX-2.3-PinkCherry" has no
slash, so basename() hands back the whole prefixed string. Every per-model
image / video / audio / tts / embedding setting was therefore replaced by the
global default, silently, for any model addressed by its alias -- which is the
only name that distinguishes two configs sharing one checkpoint path. The
resolver now tries the typed keys, ordered by the types the model is actually
registered as. images.py (6 sites) and tts.py looked their config up with a
raw config.get(model_key) and had the same bug; they resolve now, with a test
that fails if a raw one reappears.

The flag itself lived only in _raw_cfg, where a consumer reading
config['sequential_text_encoder'] never saw it. Promoted to a top-level
runtime kwarg, as gpu_split beside it already documents.

With the gate finally firing, the render failed: "Must provide
`prompt_attention_mask` when specifying `prompt_embeds`". encode_prompt's
return order had been guessed. Read off the installed sources, LTX-2 returns
(embeds, mask, neg_embeds, neg_mask) and Krea 2 returns (embeds, mask) --
the mask comes SECOND, and the old code read it third, filing negative
embeddings as the attention mask. Naming is now done by classifying the
values (a mask is an integer tensor, embeddings are floating) instead of by
position.

That rewrite then dropped Krea 2's negative prompt, which would have been
worse than the failure it replaced: Krea 2's __call__ encodes "" itself when
handed no negative, against an encoder this function has just moved to the
CPU. The negative is always encoded up front for those families now.

Hardened for families nobody has tried: an SDXL-shaped pipeline returns four
float tensors including the pooled pair, and naming only two does not degrade
quietly -- the pipeline rejects incomplete embeddings and the request fails.
If everything encode_prompt returned cannot be named, the optimisation is
abandoned for that request and says so.

Measured on LTX-2.3-PinkCherry (480x480, 25 frames, 8 steps, RTX 3090):

    without   369.4 s   peak 23718 MiB   91% util
    with      345.3 s   peak 24098 MiB   92% util   7.3 GiB freed

Read as headroom, not speed: the card is 24 GiB and the denoiser fills what
it is given, so peak sits at the ceiling either way and the 24 s is one
sample. The 7.3 GiB is what stops these renders being pushed to CPU offload,
where the same job took tens of minutes.

Also here: a clip-audio scene no longer needs an explicit duration (B1 said
the duration comes from the clip, and validation refused exactly the request
the feature exists for), and _free now reports per component whether it moved,
was already off the GPU, or REFUSED the move -- a bnb-quantized encoder can
refuse, and all three used to arrive as "nothing was freed".

test_sequential_text_encoder's fakes returned bare strings in the wrong tuple
order, which is why the bug shipped: the test encoded the mistaken belief
rather than the pipeline's contract. They carry dtypes and the real orders
now. test_compose's engine-selection test patches shutil.which as well,
because the production image ships wav2lip and sadtalker shims on PATH and the
test was asserting the absence of a binary the real deployment has.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014T3rZozi1Ehk96NG1WesgC

codai/api/compose.py

@@ -503,10 +503,21 @@ def _validate(req: ComposeRequest) -> None:
                                              f"scenes[{i}].speakers[{k}]")
         has_narration = (bool((sc.text or "").strip()) or bool(sc.audio)
                          or bool(sc.speakers))
-        if not has_narration and not (sc.duration and sc.duration > 0):
+        # A scene whose sound IS its clip needs no narration and no explicit
+        # duration: `audio_mix="clip"` means "the clip's sound only, and the
+        # scene lasts as long as the clip", which _run_job already derives from
+        # the resolved media. Requiring a duration here refused exactly the
+        # request the feature exists for -- a news moment played with its own
+        # sound -- and the caller cannot always know the clip's length anyway,
+        # which is the reason the server probes it.
+        _carries_clip_audio = (_scene_mix(sc) in ("clip", "both")
+                               and _scene_clip_visual(sc) is not None)
+        if (not has_narration and not _carries_clip_audio
+                and not (sc.duration and sc.duration > 0)):
             raise HTTPException(status_code=400, detail=(
                 f"scenes[{i}]: a scene with no text and no audio needs an explicit "
-                f"\"duration\""))
+                f"\"duration\" (or a visual whose own audio it keeps, with "
+                f"audio_mix=\"clip\")"))
         pres = presenter_for(sc, req)
         if pres is not None:
             _validate_presenter(pres, f"scenes[{i}].presenter" if sc.presenter else "presenter")

codai/api/images.py

@@ -2408,15 +2408,21 @@ async def create_image_generation(request: ImageGenerationRequest, http_request:
         if pipeline is not None:
             if is_sdcpp:
                 print(f"Using cached sd.cpp model for generation")
-                _sdcpp_cfg = (multi_model_manager.config.get(model_key)
-                              or multi_model_manager.config.get(model_name) or {})
+                # Resolved, not looked up raw: typed entries are keyed
+                # "image:<name>", so a bare config.get(alias) misses and every
+                # per-model setting is silently replaced by the global default.
+                _sdcpp_cfg = (multi_model_manager._config_for_model(model_key)
+                              or multi_model_manager._config_for_model(model_name)
+                              or {})
                 return await _generate_with_sdcpp(pipeline, request, global_args,
                                                   http_request, model_config=_sdcpp_cfg)
             else:
                 # Assume it's a diffusers pipeline
                 print(f"Using cached diffusers pipeline for generation")
-                _diff_cfg = (multi_model_manager.config.get(model_key)
-                             or multi_model_manager.config.get(model_name) or {})
+                # See the note above: resolved, not a raw config.get.
+                _diff_cfg = (multi_model_manager._config_for_model(model_key)
+                             or multi_model_manager._config_for_model(model_name)
+                             or {})
                 return await _generate_with_diffusers(pipeline, request, global_args,
                                                       http_request, model_config=_diff_cfg)
         
@@ -2438,8 +2444,10 @@ async def create_image_generation(request: ImageGenerationRequest, http_request:
         if not is_gguf:
             try:
                 print(f"Loading diffusers model: {model_name}")
-                _diff_cfg = (multi_model_manager.config.get(model_key)
-                             or multi_model_manager.config.get(model_name) or {})
+                # See the note above: resolved, not a raw config.get.
+                _diff_cfg = (multi_model_manager._config_for_model(model_key)
+                             or multi_model_manager._config_for_model(model_name)
+                             or {})
                 _vram_before = multi_model_manager.vram_before_load()
                 # Reserve the estimate for the whole load window, the way the
                 # video path does. Without it an image load is INVISIBLE to the
@@ -2498,7 +2506,8 @@ async def create_image_generation(request: ImageGenerationRequest, http_request:
 
             # Only use sd.cpp if we have a local file path
             if resolved_path and os.path.isfile(resolved_path):
-                cfg = multi_model_manager.config.get(model_key) or multi_model_manager.config.get(model_name) or {}
+                cfg = (multi_model_manager._config_for_model(model_key)
+                       or multi_model_manager._config_for_model(model_name) or {})
                 _vram_before = multi_model_manager.vram_before_load()
                 multi_model_manager.note_loading(model_key)
                 try:
@@ -2817,8 +2826,8 @@ async def _pipeline_for_task(model_info: dict, task: str):
     configured_error = None
     cfg = {}
     if base is None:
-        cfg = (multi_model_manager.config.get(model_key)
-               or multi_model_manager.config.get(model_name) or {})
+        cfg = (multi_model_manager._config_for_model(model_key)
+               or multi_model_manager._config_for_model(model_name) or {})
         try:
             base = await asyncio.to_thread(
                 _load_diffusers_pipeline, model_name, global_args, cfg)
@@ -3195,8 +3204,8 @@ async def create_image_inpaint(request: ImageInpaintRequest, http_request: Reque
         quality = request.quality or "standard"
         num_steps = request.steps or (30 if quality == "standard" else 50)
         cfg_scale = request.guidance_scale or (getattr(global_args, 'image_cfg_scale', 7.5) if quality == "standard" else 9.0)
-        _inp_cfg = (multi_model_manager.config.get(model_key)
-                    or multi_model_manager.config.get(model_name) or {})
+        _inp_cfg = (multi_model_manager._config_for_model(model_key)
+                    or multi_model_manager._config_for_model(model_name) or {})
         result = await asyncio.get_event_loop().run_in_executor(
             None,
             lambda: pipe(

codai/api/tts.py

@@ -182,8 +182,14 @@ async def create_speech(request: TTSRequest, http_request: Request = None):
                 if model_name.startswith(('http://', 'https://')):
                     from codai.models.cache import load_model
                     model_path = load_model(model_name) or model_name
-                cfg = multi_model_manager.config.get(model_key) or \
-                    multi_model_manager.config.get(f"tts:{model_name}") or {}
+                # Resolved rather than looked up raw: a typed entry is keyed
+                # "tts:<name>", and an ALIAS gets its own such key, so a bare
+                # get() on the request id misses and drops every per-model
+                # setting. The explicit tts: form below covers only the path.
+                cfg = (multi_model_manager._config_for_model(model_key)
+                       or multi_model_manager._config_for_model(model_name)
+                       or multi_model_manager.config.get(f"tts:{model_name}")
+                       or {})
                 with loading_task(model_name, model_type="tts"):
                     tts_backend = await asyncio.to_thread(
                         tts_backends.load_backend, model_name, model_path, cfg)

codai/api/video.py

@@ -3058,6 +3058,12 @@ def _generate_video(pipe, request: VideoGenerationRequest):
                     getattr(request, 'model', None) or '') or {}
                 _teo_raw = _teo_cfg.get('_raw_cfg') \
                     if isinstance(_teo_cfg.get('_raw_cfg'), dict) else {}
+                # The flag is read from the top-level kwargs AND from the raw
+                # card. Both, because reading only one is what kept this inert:
+                # the kwargs lookup returned {} for an aliased model until
+                # _config_for_model learned the typed keys, and the flag itself
+                # lived only in _raw_cfg until it was promoted. Either half
+                # failing silently looks exactly like "the user did not set it".
                 if bool(_teo_cfg.get('sequential_text_encoder')
                         or (_teo_raw or {}).get('sequential_text_encoder')):
                     try:
@@ -3066,6 +3072,13 @@ def _generate_video(pipe, request: VideoGenerationRequest):
                     except Exception as _teo:
                         print(f"  [text-encoder-offload] skipped ({_teo})",
                               flush=True)
+                else:
+                    # Says which model, so "nothing happened" is distinguishable
+                    # from "the flag was not found on the card it was set on".
+                    print(f"  [text-encoder-offload] not enabled for "
+                          f"{getattr(request, 'model', '?')} "
+                          f"(set `sequential_text_encoder` on its card)",
+                          flush=True)
                 frames = _run_pipeline(pipe, kw, _gen_audio)
     except TaskCancelled:
         _vid_progress_done()

codai/main.py

@@ -157,6 +157,12 @@ def build_runtime_kwargs(model_cfg, model_type):
         'tensor_split':    model_cfg.get('tensor_split'),
         'split_strategy':  model_cfg.get('split_strategy'),
         'split_secondary_cap_gb': model_cfg.get('split_secondary_cap_gb'),
+        # Promoted to a TOP-LEVEL kwarg for the same reason as gpu_split above:
+        # a consumer reading config['...'] never sees a key that only exists in
+        # _raw_cfg. Measured: the card said sequential_text_encoder=True and the
+        # video path still logged "not enabled", because it was looking at the
+        # runtime kwargs and the flag lived one level down.
+        'sequential_text_encoder': bool(model_cfg.get('sequential_text_encoder', False)),
         '_raw_cfg':        dict(model_cfg) if isinstance(model_cfg, dict) else {},
     }
     if model_type == "text":

codai/models/manager.py

@@ -2636,6 +2636,28 @@ class MultiModelManager:
         target = self.model_aliases.get(name)
         if target and target != name and self.config.get(target):
             return target
+        # Typed entries are keyed "<kind>:<name>" (see _store_type_config), so a
+        # plain lookup by alias or path misses every non-text model -- and the
+        # basename fallback below misses too, because "video:LTX-2.3-PinkCherry"
+        # has no slash and basename() returns the WHOLE prefixed string.
+        #
+        # Measured 2026-10-28: _config_for_model('LTX-2.3-PinkCherry') returned
+        # {} on a card that plainly had sequential_text_encoder=True, so the
+        # feature logged "not enabled" and no per-model video/image setting was
+        # reachable by the only name that distinguishes a multi-config sibling.
+        # The kinds this name is actually registered as come first, so a path
+        # registered under two types resolves to one of ITS types rather than
+        # whichever the fixed order happened to name.
+        _kinds = []
+        try:
+            _kinds = sorted(self._registered_types_for(name) or ())
+        except Exception:
+            _kinds = []
+        for _kind in _kinds + ["text", "image", "video", "audio", "tts",
+                               "vision", "audio_gen", "embedding", "spatial"]:
+            for _n in (name, target):
+                if _n and self.config.get(f"{_kind}:{_n}"):
+                    return f"{_kind}:{_n}"
         import os
         base = os.path.basename(str(name))
         base_noext = base[:-5] if base.endswith(".gguf") else base

tests/test_compose.py

@@ -1026,6 +1026,13 @@ def test_a_character_profile_supplies_the_face(tmp_path, files_dir, monkeypatch)
 
 def test_engine_selection_prefers_video_engines_for_a_clip(monkeypatch):
     from codai.compose import presenter as pres
+    # shutil.which is patched alongside available_engines because pick_engine
+    # accepts a named engine that is merely ON PATH. The production image ships
+    # wav2lip and sadtalker shims in /usr/local/bin, so without this the
+    # "not installed" cases cannot be expressed and the test fails there while
+    # passing on a bare host -- it was asserting the absence of a binary that
+    # the real deployment has.
+    monkeypatch.setattr(pres.shutil, "which", lambda _n: None)
     monkeypatch.setattr(pres, "available_engines", lambda: ["wav2lip", "sadtalker"])
     assert pres.pick_engine("auto", has_base=True) == "wav2lip"
     assert pres.pick_engine("auto", has_base=False) == "sadtalker"

tests/test_compose_clip_audio.py

@@ -287,3 +287,54 @@ def test_the_picture_path_is_never_replaced_by_the_audio_file():
     # The visual cache must hand back `path`, not the audio file.
     assert '_cached["path"]' in src, (
         "the visual no longer reuses the resolved picture path")
+
+
+# ------------------------- a clip-audio scene needs no explicit duration
+#
+# B1 says audio_mix="clip" means "the clip's sound only (no narration, scene
+# duration = the clip's)". Validation still demanded an explicit duration for
+# any scene with no text and no audio, so it refused exactly the request the
+# feature exists for -- a news moment played with its own sound. Found by
+# submitting one, not by the unit tests, which exercise _scene_mix and
+# _scene_clip_visual in isolation and never reach _validate.
+
+def test_a_clip_audio_scene_needs_no_explicit_duration():
+    from codai.api.compose import ComposeRequest, _validate
+    req = ComposeRequest(scenes=[Scene(
+        text="", audio_mix="clip",
+        visuals=[Visual(type="video", src="/v1/files/x.mp4", keep_audio=True)])])
+    _validate(req)          # must not raise
+
+
+def test_audio_mix_both_also_counts_as_carrying_sound():
+    from codai.api.compose import ComposeRequest, _validate
+    req = ComposeRequest(scenes=[Scene(
+        text="", audio_mix="both",
+        visuals=[Visual(type="video", src="/v1/files/x.mp4", keep_audio=True)])])
+    _validate(req)
+
+
+def test_a_silent_scene_still_needs_a_duration():
+    """The check is not simply removed: a scene with nothing to time by is
+    still a mistake worth catching before the render starts."""
+    import pytest as _pytest
+    from fastapi import HTTPException
+    from codai.api.compose import ComposeRequest, _validate
+    req = ComposeRequest(scenes=[Scene(
+        text="", visuals=[Visual(type="color", color="#000000")])])
+    with _pytest.raises(HTTPException) as e:
+        _validate(req)
+    assert e.value.status_code == 400
+    assert "duration" in str(e.value.detail)
+
+
+def test_a_clip_with_no_keep_audio_still_needs_a_duration():
+    """audio_mix alone is not enough -- nothing is marked as the sound source."""
+    import pytest as _pytest
+    from fastapi import HTTPException
+    from codai.api.compose import ComposeRequest, _validate
+    req = ComposeRequest(scenes=[Scene(
+        text="", audio_mix="clip",
+        visuals=[Visual(type="video", src="a.mp4"), Visual(type="video", src="b.mp4")])])
+    with _pytest.raises(HTTPException):
+        _validate(req)

tests/test_sequential_text_encoder.py

@@ -39,6 +39,33 @@ import pytest
 from codai.models import text_encoder_offload as T
 
 
+class _T:
+    """A stand-in for a returned tensor.
+
+    It carries a dtype because that is how the embeddings are told apart from
+    the attention mask -- the ORDER differs per family and cannot be assumed.
+    A fake that returned bare strings could not exercise that at all, which is
+    how the real tuple order went unchecked: this file's _LtxLike used to
+    return (embeds, neg_embeds, mask, neg_mask), the live pipeline returns
+    (embeds, mask, neg_embeds, neg_mask), and the live render failed with
+    "Must provide `prompt_attention_mask` when specifying `prompt_embeds`".
+    """
+
+    class _D:
+        def __init__(self, f):
+            self.is_floating_point = f
+
+    def __init__(self, tag, mask=False):
+        self.tag = tag
+        self.dtype = _T._D(not mask)
+
+    def __eq__(self, other):
+        return isinstance(other, str) and other == self.tag or self is other
+
+    def __repr__(self):
+        return f"<{self.tag}>"
+
+
 class _KreaLike:
     """Two-call family: the negative prompt needs its own encode."""
     text_encoder = object()
@@ -49,7 +76,7 @@ class _KreaLike:
 
     def encode_prompt(self, prompt, device=None, max_sequence_length=512):
         self.calls.append(prompt)
-        return (f"E:{prompt}", f"M:{prompt}")
+        return (_T(f"E:{prompt}"), _T(f"M:{prompt}", mask=True))
 
     def __call__(self, prompt=None, negative_prompt=None, prompt_embeds=None,
                  prompt_embeds_mask=None, negative_prompt_embeds=None,
@@ -69,7 +96,11 @@ class _LtxLike:
                       num_videos_per_prompt=1):
         self.calls.append((prompt, negative_prompt,
                            do_classifier_free_guidance))
-        return ("E", "NE", "M", "NM")
+        # The real order, copied from diffusers' pipeline_ltx2.py:
+        #   prompt_embeds, prompt_attention_mask,
+        #   negative_prompt_embeds, negative_prompt_attention_mask
+        return (_T("E"), _T("M", mask=True),
+                _T("NE"), _T("NM", mask=True))
 
     def __call__(self, prompt=None, negative_prompt=None, prompt_embeds=None,
                  prompt_attention_mask=None, negative_prompt_embeds=None,

tests/test_text_encoder_offload_outputs.py

+# CoderAI - OpenAI-compatible API server
+# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+"""The embeddings must be paired with the right mask.
+
+``encode_prompt`` returns a bare tuple whose order differs per family and is
+not discoverable from the signature. Read from the wrong slot, the pipeline
+gets a negative-embeddings tensor where its attention mask belongs.
+
+Measured on a live LTX-2.3 render: the request failed with "Must provide
+`prompt_attention_mask` when specifying `prompt_embeds`" because the code
+assumed (embeds, neg_embeds, mask, neg_mask) while diffusers returns
+(embeds, mask, neg_embeds, neg_mask). The orders asserted here are copied
+from the installed diffusers sources, so a family added later has to state
+its own.
+"""
+
+import os
+import sys
+
+import pytest
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+torch = pytest.importorskip("torch")
+
+from codai.models import text_encoder_offload as teo
+
+
+def _embeds(n=3):
+    return torch.zeros((1, n, 8), dtype=torch.float32)
+
+
+def _mask(n=3):
+    return torch.ones((1, n), dtype=torch.int64)
+
+
+LTX_PARAMS = {"prompt_embeds", "prompt_attention_mask",
+              "negative_prompt_embeds", "negative_prompt_attention_mask"}
+KREA_PARAMS = {"prompt_embeds", "prompt_embeds_mask",
+               "negative_prompt_embeds", "negative_prompt_embeds_mask"}
+
+
+def test_ltx2_order_is_embeds_mask_negembeds_negmask():
+    """diffusers/pipelines/ltx2/pipeline_ltx2.py returns exactly this."""
+    pe, pm, ne, nm = _embeds(), _mask(), _embeds(5), _mask(5)
+    got = teo._assign_outputs((pe, pm, ne, nm), LTX_PARAMS,
+                              "prompt_attention_mask")
+    assert got["prompt_embeds"] is pe
+    assert got["prompt_attention_mask"] is pm, \
+        "the mask must come from slot 1, not slot 2"
+    assert got["negative_prompt_embeds"] is ne
+    assert got["negative_prompt_attention_mask"] is nm
+
+
+def test_krea2_order_is_embeds_mask():
+    """diffusers/pipelines/krea2/pipeline_krea2.py returns a 2-tuple."""
+    pe, pm = _embeds(), _mask()
+    got = teo._assign_outputs((pe, pm), KREA_PARAMS, "prompt_embeds_mask")
+    assert got["prompt_embeds"] is pe
+    assert got["prompt_embeds_mask"] is pm
+    assert "negative_prompt_embeds" not in got
+
+
+def test_a_family_with_no_mask_is_not_handed_embeddings_as_one():
+    """Two float tensors = positive and negative embeddings, no mask."""
+    pe, ne = _embeds(), _embeds(5)
+    got = teo._assign_outputs((pe, ne), LTX_PARAMS, "prompt_attention_mask")
+    assert got["prompt_embeds"] is pe
+    assert got.get("negative_prompt_embeds") is ne
+    assert "prompt_attention_mask" not in got, \
+        "a float tensor was misfiled as an attention mask"
+
+
+def test_none_entries_are_skipped():
+    """LTX returns None negatives when guidance is off."""
+    pe, pm = _embeds(), _mask()
+    got = teo._assign_outputs((pe, pm, None, None), LTX_PARAMS,
+                              "prompt_attention_mask")
+    assert got["prompt_attention_mask"] is pm
+    assert "negative_prompt_embeds" not in got
+
+
+def test_unsupported_kwargs_are_not_passed_on():
+    """A name the pipeline does not accept must not reach the call."""
+    pe, pm = _embeds(), _mask()
+    got = teo._assign_outputs((pe, pm), {"prompt_embeds"},
+                              "prompt_embeds_mask")
+    assert got == {"prompt_embeds": pe}
+
+
+def test_missing_mask_aborts_rather_than_failing_the_request():
+    """If the mask cannot be paired, generate the ordinary way instead of
+    handing the pipeline embeddings it will reject."""
+
+    class _Pipe:
+        text_encoder = object()
+
+        def __call__(self, prompt=None, prompt_embeds=None,
+                     prompt_attention_mask=None):
+            raise AssertionError("must not be called")
+
+        def encode_prompt(self, prompt=None, negative_prompt=None):
+            # A family that returns embeddings only -- no mask to pair.
+            return (_embeds(),)
+
+    pipe = _Pipe()
+    kwargs = {"prompt": "a cat"}
+    out, freed = teo.encode_then_free(pipe, kwargs)
+    assert out is kwargs or out == kwargs, \
+        "the original kwargs must come back untouched"
+    assert freed == 0.0
+
+
+def test_sdxl_shape_is_named_in_full_not_partially():
+    """SDXL returns (embeds, neg_embeds, pooled, neg_pooled) -- four floats and
+    no mask. Naming only the first two and dropping the pooled pair makes the
+    pipeline raise "Must provide pooled_prompt_embeds", so the whole shape has
+    to be named or nothing is."""
+    sdxl_params = {"prompt_embeds", "negative_prompt_embeds",
+                   "pooled_prompt_embeds", "negative_pooled_prompt_embeds"}
+    pe, ne, po, npo = _embeds(), _embeds(5), _embeds(1), _embeds(1)
+    got = teo._assign_outputs((pe, ne, po, npo), sdxl_params, None)
+    assert got["prompt_embeds"] is pe
+    assert got["negative_prompt_embeds"] is ne
+    assert got["pooled_prompt_embeds"] is po
+    assert got["negative_pooled_prompt_embeds"] is npo
+
+
+def test_an_unnameable_shape_refuses_rather_than_dropping_tensors():
+    """A family returning more than we can name must abandon the optimisation,
+    not hand the pipeline an incomplete set it will reject."""
+    many = tuple(_embeds() for _ in range(7))
+    with pytest.raises(ValueError, match="could be named"):
+        teo._assign_outputs(many, {"prompt_embeds"}, None)
+
+
+def test_an_unnameable_shape_falls_back_to_the_ordinary_path():
+    """End to end: the request still generates, via prompt rather than
+    embeddings."""
+
+    class _Pipe:
+        text_encoder = object()
+
+        def __call__(self, prompt=None, prompt_embeds=None):
+            raise AssertionError("must not be called by the helper")
+
+        def encode_prompt(self, prompt=None):
+            return tuple(_embeds() for _ in range(7))
+
+    kwargs = {"prompt": "a cat"}
+    out, freed = teo.encode_then_free(_Pipe(), kwargs)
+    assert out == kwargs, "the prompt must survive for the ordinary path"
+    assert "prompt_embeds" not in out
+    assert freed == 0.0
+
+
+class _KreaLike:
+    """A separate-negative family: encode_prompt has no `negative_prompt`, so
+    the negative needs its own call. Mirrors Krea 2, whose __call__ otherwise
+    encodes "" itself at denoise time -- against an encoder we just moved."""
+
+    text_encoder = object()
+
+    def __init__(self):
+        self.prompts = []
+
+    def __call__(self, prompt=None, prompt_embeds=None, prompt_embeds_mask=None,
+                 negative_prompt=None, negative_prompt_embeds=None,
+                 negative_prompt_embeds_mask=None):
+        raise AssertionError("must not be called by the helper")
+
+    def encode_prompt(self, prompt=None, device=None, max_sequence_length=None):
+        self.prompts.append(prompt)
+        return _embeds(), _mask()
+
+
+def test_separate_negative_family_gets_its_negative_encoded():
+    pipe = _KreaLike()
+    out, _ = teo.encode_then_free(pipe, {"prompt": "a cat",
+                                         "negative_prompt": "blurry"})
+    assert pipe.prompts == ["a cat", "blurry"], \
+        f"the negative prompt was not encoded: {pipe.prompts}"
+    assert "negative_prompt_embeds" in out
+    assert "negative_prompt_embeds_mask" in out
+    assert "negative_prompt" not in out and "prompt" not in out
+
+
+def test_separate_negative_family_encodes_empty_when_none_given():
+    """With no negative supplied the pipeline would encode "" itself, at
+    denoise time, on an encoder that is no longer on the GPU."""
+    pipe = _KreaLike()
+    out, _ = teo.encode_then_free(pipe, {"prompt": "a cat"})
+    assert pipe.prompts == ["a cat", ""], \
+        f"empty negative was not encoded up front: {pipe.prompts}"
+    assert out.get("negative_prompt_embeds") is not None
+
+
+def test_both_in_one_family_is_not_called_twice():
+    """LTX-2 encodes both in a single call; a second one would be waste."""
+
+    calls = []
+
+    class _LtxLike:
+        text_encoder = object()
+
+        def __call__(self, prompt=None, prompt_embeds=None,
+                     prompt_attention_mask=None,
+                     negative_prompt_embeds=None,
+                     negative_prompt_attention_mask=None):
+            raise AssertionError("must not be called by the helper")
+
+        def encode_prompt(self, prompt=None, negative_prompt=None,
+                          do_classifier_free_guidance=True, device=None):
+            calls.append(prompt)
+            return _embeds(), _mask(), _embeds(5), _mask(5)
+
+    out, _ = teo.encode_then_free(_LtxLike(), {"prompt": "a cat",
+                                               "negative_prompt": "blurry",
+                                               "guidance_scale": 3.0})
+    assert calls == ["a cat"], f"encoded more than once: {calls}"
+    assert out["negative_prompt_attention_mask"] is not None

tests/test_typed_config_lookup.py

+# CoderAI - OpenAI-compatible API server
+# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+"""A non-text model must be reachable by its alias.
+
+Typed entries are keyed ``<kind>:<name>``, so every lookup form that works for
+a text model (plain path, plain basename) misses for a video or image one. The
+basename fallback misses as well, because ``video:LTX-2.3-PinkCherry`` contains
+no slash and ``basename()`` hands back the whole prefixed string.
+
+The consequence was silent: ``_config_for_model`` returned ``{}`` for a card
+that carried the setting, so per-model video/image settings were simply not
+applied and the feature reported itself "not enabled". Two sibling configs
+sharing one checkpoint path are only distinguishable BY alias, so this is the
+one lookup that has to work for them.
+"""
+
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+
+class _Resolver:
+    """The two methods under test, lifted onto the real class."""
+
+    def __init__(self, config, aliases):
+        self.config = config
+        self.model_aliases = aliases
+
+    @staticmethod
+    def _bind():
+        from codai.models.manager import MultiModelManager
+        _Resolver._resolve_config_key = MultiModelManager._resolve_config_key
+        _Resolver._config_for_model = MultiModelManager._config_for_model
+
+
+_Resolver._bind()
+
+
+def _two_siblings():
+    """Two video configs sharing one checkpoint path, as models.json has them."""
+    path = "diffusers/LTX-2.3-Diffusers"
+    eros = {"alias": "LTX-2.3-10Eros", "path": path,
+            "sequential_text_encoder": True, "marker": "eros"}
+    pink = {"alias": "LTX-2.3-PinkCherry", "path": path,
+            "sequential_text_encoder": True, "marker": "pink"}
+    config = {
+        # _store_type_config writes the path key (last writer wins) and an
+        # alias key per entry.
+        f"video:{path}": pink,
+        "video:LTX-2.3-10Eros": eros,
+        "video:LTX-2.3-PinkCherry": pink,
+    }
+    aliases = {"LTX-2.3-10Eros": path, "LTX-2.3-PinkCherry": path}
+    return _Resolver(config, aliases)
+
+
+def test_alias_resolves_to_its_own_typed_entry():
+    r = _two_siblings()
+    for alias, marker in (("LTX-2.3-10Eros", "eros"),
+                          ("LTX-2.3-PinkCherry", "pink")):
+        cfg = r._config_for_model(alias)
+        assert cfg, f"{alias} resolved to nothing -- every setting is dropped"
+        assert cfg.get("sequential_text_encoder") is True, \
+            f"{alias} lost its per-model setting"
+        assert cfg.get("marker") == marker, \
+            f"{alias} resolved to the sibling's config ({cfg.get('marker')})"
+
+
+def test_path_still_resolves():
+    """The path key keeps its meaning for callers that know only a path."""
+    r = _two_siblings()
+    cfg = r._config_for_model("diffusers/LTX-2.3-Diffusers")
+    assert cfg.get("marker") == "pink"  # last writer, as before
+
+
+def test_unknown_name_still_returns_empty():
+    r = _two_siblings()
+    assert r._config_for_model("no-such-model") == {}
+    assert r._config_for_model("") == {}
+
+
+def test_exact_key_is_preferred_over_the_typed_scan():
+    """An unprefixed entry must still win, so text models are unaffected."""
+    r = _Resolver({"plain-model": {"marker": "plain"},
+                   "text:plain-model": {"marker": "typed"}}, {})
+    assert r._config_for_model("plain-model").get("marker") == "plain"
+
+
+def test_flag_is_a_top_level_runtime_kwarg():
+    """The gate reads config['sequential_text_encoder'], so it must be promoted
+    out of _raw_cfg -- the same trap already documented for gpu_split."""
+    from codai.main import build_runtime_kwargs
+    on = build_runtime_kwargs({"sequential_text_encoder": True}, "video")
+    off = build_runtime_kwargs({}, "video")
+    assert on["sequential_text_encoder"] is True
+    assert off["sequential_text_encoder"] is False
+
+
+def test_the_image_and_tts_paths_resolve_instead_of_a_raw_get():
+    """Three API paths looked their config up with a bare
+    ``multi_model_manager.config.get(model_key)``. Typed entries are keyed
+    "<kind>:<name>", so that get() misses for every alias and the per-model
+    settings are replaced by global defaults with nothing logged. The video
+    path's sequential_text_encoder reporting itself "not enabled" was this bug
+    one module over, so the same lookups elsewhere are pinned here."""
+    import re
+    root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+    for rel in ("codai/api/images.py", "codai/api/tts.py"):
+        body = open(os.path.join(root, rel), encoding="utf-8").read()
+        bad = re.findall(r"config\.get\(model_key\)", body)
+        assert not bad, (
+            f"{rel} looks its model config up with a raw config.get(model_key); "
+            f"use _config_for_model so a typed/alias key resolves")