three image failures that each reported something other than the cause

From the model test of 2026-11-01.

A mask handed over as a negative. encode_prompt returns a 2-tuple for several
families and nothing about its LENGTH says which: SD 1.x means
(prompt_embeds, negative_prompt_embeds), Krea 2 and Qwen-Image mean
(prompt_embeds, prompt_embeds_mask), MOSS-Image and Kromoss mean (embeds, mask)
and name the parameter prompt_attention_mask. The disambiguation looked for one
name -- the Krea spelling -- so MOSS fell through to the SD 1.x reading and its
boolean [B, S] mask was passed as negative_prompt_embeds. Every MOSS generation
then failed with "Prompt embeddings must be [B, S, C] with matching batch
size.": true, and about an argument the caller never meant to send. Now the
signature is asked for both spellings and, when it is silent, the tensors
decide -- a negative matches the positive in rank and width, a mask is boolean
or one rank short. The test that pinned the old check was called "told apart by
signature not by name" while asserting a single name, which is how the gap
survived.

An error with nothing after it. str(exc) is empty for any exception raised with
no arguments, and interpolating it produced "Image generation failed ... so this
is not a load failure." followed by nothing -- measured on Z-Image-Turbo, where
the reader could not tell a shape error from a full card. _exc_text always
names the type and says plainly when there is no message.

A promise the loader cannot keep. component_quantization's docstring said a
single-file component could be quantized as it loads. It cannot:
from_single_file given a bitsandbytes config takes the PRE-quantized path, so a
plain community fine-tune fails after the whole file is read, naming a tensor
instead of the cause. Refused up front now -- before the class lookup, which
`continue`s when the installed diffusers has no class for the component and so
skipped the check entirely -- naming the two routes that do work.

Also: the text-only image refusal on /v1/embeddings was gated on
`model_obj is None`, so the same request was refused 400 on a cold server and
admitted on a warm one, where only the loaded-object check stood between images
and a text-only encoder. A correctness check must not read a cache state.
Co-Authored-By: 's avatarClaude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014T3rZozi1Ehk96NG1WesgC
parent b86e7e94
...@@ -2730,7 +2730,13 @@ async def _run_embeddings(request: EmbeddingsRequest, http_request: Request = No ...@@ -2730,7 +2730,13 @@ async def _run_embeddings(request: EmbeddingsRequest, http_request: Request = No
f"those. Send an image, or use a text/multimodal " f"those. Send an image, or use a text/multimodal "
f"embedding model.") f"embedding model.")
if images and model_obj is None and not _may_embed_images(model_name, _emb_cfg): # NOT gated on `model_obj is None`. It was, which made a refusal depend on
# whether the model happened to be resident: the same request was answered
# 400 on a cold server and admitted on a warm one, where only the
# loaded-object check below stood between it and the encoder. A correctness
# check must not read a cache state. The cost of asking twice is one
# config.json lookup, which :func:`_may_embed_images` reads from disk.
if images and not _may_embed_images(model_name, _emb_cfg):
raise HTTPException( raise HTTPException(
status_code=400, status_code=400,
detail=f"Model '{model_name}' is text-only; image embedding needs a " detail=f"Model '{model_name}' is text-only; image embedding needs a "
......
...@@ -670,6 +670,75 @@ _CUDA_OOM_MARKERS = ( ...@@ -670,6 +670,75 @@ _CUDA_OOM_MARKERS = (
) )
def _exc_text(exc) -> str:
"""An exception rendered so the reader always learns something.
``str(exc)`` is empty for any exception raised with no arguments, and
several of the ones that reach this path are: a bare ``RuntimeError()`` out
of a pipeline's own validation, an ``AssertionError`` from a shape check.
Interpolated straight into a sentence it produces a message that simply
stops -- measured 2026-11-01, where Z-Image-Turbo answered "Image
generation failed ... so this is not a load failure." with nothing after it,
and the reader could not tell a shape error from a full card.
So always name the type, and say plainly when there is no message rather
than leaving a blank where one should be.
"""
text = str(exc or "").strip()
name = type(exc).__name__
if not text:
return f"{name} (raised with no message)"
if text.startswith(name):
return text
return f"{name}: {text}"
def _second_is_a_mask(enc, call_params) -> bool:
"""Whether a 2-tuple from ``encode_prompt`` is (embeds, MASK), not (pos, neg).
Both shapes exist and they are indistinguishable by length. SD 1.x returns
``(prompt_embeds, negative_prompt_embeds)``; Krea 2 and the Qwen-Image
family return ``(prompt_embeds, prompt_embeds_mask)``; MOSS-Image and
Kromoss return ``(embeds, mask)`` and call the parameter
``prompt_attention_mask``.
This used to be decided by looking for the single name
``prompt_embeds_mask``, so MOSS fell through to the SD 1.x branch and its
boolean ``[B, S]`` mask was handed over as ``negative_prompt_embeds``. The
pipeline then said
Prompt embeddings must be [B, S, C] with matching batch size.
-- true, and about an argument the caller never meant to pass. Measured
2026-11-01, every MOSS-Image generation.
So ask the tensors, not the vocabulary: a negative embedding matches the
positive in rank and width, and a mask is boolean or one rank short. The
names are still consulted first because they are free and unambiguous when
present.
"""
if {"prompt_embeds_mask", "prompt_attention_mask"} & set(call_params or ()):
return True
try:
pos, second = enc[0], enc[1]
except Exception:
return False
try:
if "bool" in str(getattr(second, "dtype", "")):
return True
pos_ndim = getattr(pos, "ndim", None)
snd_ndim = getattr(second, "ndim", None)
if pos_ndim is not None and snd_ndim is not None and snd_ndim != pos_ndim:
return True
if pos_ndim == 3 and second.shape[-1] != pos.shape[-1]:
return True
except Exception:
# Not tensors, or a shape that cannot be compared. The name check above
# is the only evidence, and it said nothing.
return False
return False
def _is_cuda_oom(exc) -> bool: def _is_cuda_oom(exc) -> bool:
"""Whether this exception is the GPU running out of memory. """Whether this exception is the GPU running out of memory.
...@@ -1923,13 +1992,15 @@ async def _generate_with_diffusers(pipeline, request, global_args, http_request= ...@@ -1923,13 +1992,15 @@ async def _generate_with_diffusers(pipeline, request, global_args, http_request=
# enc is a tuple; length varies by pipeline type # enc is a tuple; length varies by pipeline type
_call_params = set( _call_params = set(
_inspect.signature(pipeline.__call__).parameters) _inspect.signature(pipeline.__call__).parameters)
if len(enc) == 2 and 'prompt_embeds_mask' in _call_params: if len(enc) == 2 and _second_is_a_mask(enc, _call_params):
# Krea 2 and the Qwen-Image family also return a 2-tuple, # Krea 2, the Qwen-Image family, MOSS-Image and Kromoss all
# but it is (prompt_embeds, prompt_embeds_mask) -- not a # return a 2-tuple, but it is (prompt_embeds, mask) -- not a
# negative. Passing the mask AS the negative is what the # negative. Passing the mask AS the negative is what Krea 2
# pipeline reports as "If `prompt_embeds` are provided, # reports as "If `prompt_embeds` are provided,
# `prompt_embeds_mask` also have to be passed", which reads # `prompt_embeds_mask` also have to be passed" and what MOSS
# like a missing argument rather than a mislabelled one. # reports as "Prompt embeddings must be [B, S, C]": both read
# like a missing or malformed argument rather than a
# mislabelled one. See :func:`_second_is_a_mask`.
# These are not CLIP-pair pipelines, so weighted 75-token # These are not CLIP-pair pipelines, so weighted 75-token
# windows mean nothing to them: let the pipeline encode the # windows mean nothing to them: let the pipeline encode the
# prompt itself, the way Z-Image already does. # prompt itself, the way Z-Image already does.
...@@ -2784,11 +2855,12 @@ async def create_image_generation(request: ImageGenerationRequest, http_request: ...@@ -2784,11 +2855,12 @@ async def create_image_generation(request: ImageGenerationRequest, http_request:
raise raise
except Exception as e: except Exception as e:
import traceback import traceback
_etext = _exc_text(e)
print(f"image generation error (model '{model_name}' IS " print(f"image generation error (model '{model_name}' IS "
f"loaded): {e}") f"loaded): {_etext}")
print(f"Traceback: {traceback.format_exc()}") print(f"Traceback: {traceback.format_exc()}")
_oom = _is_cuda_oom(e) _oom = _is_cuda_oom(e)
_treg.finish(_ltid, "error", str(e)) _treg.finish(_ltid, "error", _etext)
raise HTTPException( raise HTTPException(
status_code=503 if _oom else 500, status_code=503 if _oom else 500,
detail=(f"Image generation failed for '{model_name}' — " detail=(f"Image generation failed for '{model_name}' — "
...@@ -2797,8 +2869,8 @@ async def create_image_generation(request: ImageGenerationRequest, http_request: ...@@ -2797,8 +2869,8 @@ async def create_image_generation(request: ImageGenerationRequest, http_request:
+ (f"The GPU ran out of memory while making " + (f"The GPU ran out of memory while making "
f"the image — the VAE decode and cuBLAS's " f"the image — the VAE decode and cuBLAS's "
f"own workspace are the usual points of " f"own workspace are the usual points of "
f"failure: {e}" f"failure: {_etext}"
if _oom else str(e)))) if _oom else _etext)))
# Try stable-diffusion-cpp-python (for GGUF models or as fallback) # Try stable-diffusion-cpp-python (for GGUF models or as fallback)
try: try:
......
...@@ -1105,9 +1105,24 @@ def build_component_models(model_name: str, cfg: Optional[Dict[str, Any]], dtype ...@@ -1105,9 +1105,24 @@ def build_component_models(model_name: str, cfg: Optional[Dict[str, Any]], dtype
``QwenImageEditPlusPipeline.from_pretrained(base, transformer= ``QwenImageEditPlusPipeline.from_pretrained(base, transformer=
QwenImageTransformer2DModel.from_pretrained(derivative))``. QwenImageTransformer2DModel.from_pretrained(derivative))``.
The component's own entry in ``component_quantization`` still applies, so a The component's own entry in ``component_quantization`` still applies to a
bf16 or fp8 transformer too large for the card can be quantized as it component loaded from a REPO, so a bf16 or fp8 transformer too large for the
loads: ``{"transformer": "4bit"}`` alongside the override above. card can be quantized as it loads: ``{"transformer": "4bit"}`` alongside the
override above.
It does NOT apply to a component that is a single FILE. This docstring said
it did, and it does not: ``from_single_file`` given a bitsandbytes config
takes the PRE-quantized path -- it expects the checkpoint to already carry
bnb metadata and reads none, so the load dies with
Supplied state dict for transformer_blocks.0.attn.to_gate.weight does
not contain `bitsandbytes__*` and possibly other `quantized_stats`
components
which names a tensor and not the cause. Measured 2026-11-01 on
Krea-2-RealDream (``civitai:3396368:3304483``). Asked for bnb on a single
file, this now refuses BEFORE reading 12.9 GB and names the two things that
do work: a GGUF of the same component, or no quantization plus offload.
Returns ``(components_dict, description)``; empty when none configured. Returns ``(components_dict, description)``; empty when none configured.
""" """
...@@ -1130,6 +1145,27 @@ def build_component_models(model_name: str, cfg: Optional[Dict[str, Any]], dtype ...@@ -1130,6 +1145,27 @@ def build_component_models(model_name: str, cfg: Optional[Dict[str, Any]], dtype
source = str(source or '').strip() source = str(source or '').strip()
if not source: if not source:
continue continue
mode = str(comp_q.get(name) or '').strip().lower()
if mode in ("4bit", "8bit") and is_single_file_ref(source):
# REFUSE, rather than pass a config that cannot work, and refuse
# HERE -- before the class lookup and long before the file is read.
# bnb on a single file means "this checkpoint is already quantized,
# read its metadata", and a plain community fine-tune carries none,
# so the load reads the whole 12.9 GB and then fails naming a
# tensor. Whether the base pipeline resolves a class for the
# component does not enter into it: the combination is impossible as
# configured. Say what the operator can actually do instead; see the
# docstring.
raise RuntimeError(
f"component '{name}' is a single file ({source}) and "
f"component_quantization asks for {mode}, which "
f"from_single_file cannot do: bitsandbytes there reads a "
f"checkpoint that is ALREADY quantized, and this one is not. "
f"Either point component_quantization['{name}'] at a GGUF of "
f"the same component (that path quantizes on load), or drop it "
f"and give the model enough room another way "
f"(offload_strategy, a smaller size, or a repo that publishes "
f"the component pre-quantized).")
spec = specs.get(name) spec = specs.get(name)
cls_name = spec[1] if spec and len(spec) > 1 else None cls_name = spec[1] if spec and len(spec) > 1 else None
cls = getattr(diffusers, cls_name, None) if cls_name else None cls = getattr(diffusers, cls_name, None) if cls_name else None
...@@ -1138,7 +1174,6 @@ def build_component_models(model_name: str, cfg: Optional[Dict[str, Any]], dtype ...@@ -1138,7 +1174,6 @@ def build_component_models(model_name: str, cfg: Optional[Dict[str, Any]], dtype
f"loadable class for it (got {spec}).") f"loadable class for it (got {spec}).")
continue continue
kwargs: Dict[str, Any] = {"torch_dtype": dtype} kwargs: Dict[str, Any] = {"torch_dtype": dtype}
mode = str(comp_q.get(name) or '').strip().lower()
if mode in ("4bit", "8bit"): if mode in ("4bit", "8bit"):
try: try:
from diffusers import BitsAndBytesConfig as DiffBnb from diffusers import BitsAndBytesConfig as DiffBnb
......
...@@ -205,3 +205,20 @@ def test_a_clip_architecture_without_a_vision_config_still_counts(monkeypatch): ...@@ -205,3 +205,20 @@ def test_a_clip_architecture_without_a_vision_config_still_counts(monkeypatch):
lambda name: {"model_type": "clip_text_model", lambda name: {"model_type": "clip_text_model",
"architectures": ["CLIPModel"]}) "architectures": ["CLIPModel"]})
assert E._may_embed_images("any/model", {}) is True assert E._may_embed_images("any/model", {}) is True
def test_the_text_only_refusal_does_not_depend_on_the_model_being_resident():
"""It was gated on `model_obj is None`, so the SAME request was refused 400
on a cold server and admitted on a warm one -- where only the loaded-object
check stood between images and a text-only encoder. A correctness check must
not read a cache state.
Reported 2026-10-29: four texts and four images to Qwen3-Embedding-4B
dropped the connection after ~40 s and the engine stopped answering until it
came back on its own.
"""
body = _SRC[_SRC.index("# Refuse an impossible request BEFORE loading"):]
body = body[:body.index("if model_obj is None:")]
assert "if images and not _may_embed_images(" in body, \
"the refusal must not be conditioned on residency"
assert "model_obj is None and not _may_embed_images" not in body
# 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.
"""A 2-tuple from ``encode_prompt`` is (pos, neg) for some pipelines and
(embeds, mask) for others, and nothing about its LENGTH says which.
MOSS-Image and Kromoss return ``(embeds, mask)`` and name the parameter
``prompt_attention_mask``. The disambiguation looked for one name --
``prompt_embeds_mask``, the Krea 2 / Qwen-Image spelling -- so MOSS fell
through to the SD 1.x branch and its boolean ``[B, S]`` mask was passed as
``negative_prompt_embeds``. Every MOSS generation then failed with
Prompt embeddings must be [B, S, C] with matching batch size.
which is true, and about an argument the caller never meant to send.
"""
import pytest
from codai.api import images as images_mod
class FakeTensor:
"""Only what the check reads: ndim, shape and dtype."""
def __init__(self, shape, dtype="torch.bfloat16"):
self.shape = tuple(shape)
self.ndim = len(shape)
self.dtype = dtype
# --------------------------------------------------------------------------- #
# by name, when the signature says so
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("param", ["prompt_embeds_mask", "prompt_attention_mask"])
def test_either_mask_parameter_name_is_recognised(param):
"""Krea 2 / Qwen-Image spell it one way, MOSS the other. Only the first was
known, which is the whole bug."""
enc = (FakeTensor([1, 77, 2048]), FakeTensor([1, 77], "torch.bool"))
assert images_mod._second_is_a_mask(enc, {"prompt", param}) is True
def test_sd1x_two_tuple_is_a_negative_not_a_mask():
"""(prompt_embeds, negative_prompt_embeds): same rank, same width."""
enc = (FakeTensor([1, 77, 768]), FakeTensor([1, 77, 768]))
assert images_mod._second_is_a_mask(enc, {"prompt", "negative_prompt"}) is False
# --------------------------------------------------------------------------- #
# by shape, when the signature does not
# --------------------------------------------------------------------------- #
def test_a_boolean_second_element_is_a_mask_whatever_it_is_called():
"""MOSS's mask is bool. A negative embedding never is."""
enc = (FakeTensor([1, 1024, 2560]), FakeTensor([1, 1024], "torch.bool"))
assert images_mod._second_is_a_mask(enc, {"prompt"}) is True
def test_one_rank_short_is_a_mask():
"""[B, S] against [B, S, C], with a float dtype so only the rank tells."""
enc = (FakeTensor([1, 1024, 2560]), FakeTensor([1, 1024], "torch.float32"))
assert images_mod._second_is_a_mask(enc, {"prompt"}) is True
def test_a_different_width_at_the_same_rank_is_a_mask():
enc = (FakeTensor([1, 77, 2048]), FakeTensor([1, 77, 1]))
assert images_mod._second_is_a_mask(enc, {"prompt"}) is True
def test_identical_shapes_are_a_negative():
enc = (FakeTensor([2, 333, 4096]), FakeTensor([2, 333, 4096]))
assert images_mod._second_is_a_mask(enc, {"prompt"}) is False
# --------------------------------------------------------------------------- #
# it must never raise on the way past
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("enc", [
(),
(FakeTensor([1, 77, 768]),),
("not a tensor", "nor this"),
(None, None),
])
def test_anything_unreadable_falls_through_rather_than_raising(enc):
"""This runs on the generation path of every image request; a surprise here
must not become the error the caller sees."""
assert images_mod._second_is_a_mask(enc, {"prompt"}) is False
def test_the_call_site_uses_the_helper_and_not_the_single_name():
src = open("codai/api/images.py").read()
assert "_second_is_a_mask(enc, _call_params)" in src
assert "len(enc) == 2 and 'prompt_embeds_mask' in _call_params" not in src, \
"the one-name check is what let MOSS through"
def test_a_mask_means_the_pipeline_encodes_for_itself():
"""The outcome for a mask pipeline is no embeds at all -- the pipeline's own
encoder runs, as Z-Image already does. Passing embeds plus a mask we do not
understand is how this broke."""
src = open("codai/api/images.py").read()
head = src.index("_second_is_a_mask(enc, _call_params)")
body = src[head:head + 1200]
assert "embed_kwargs = {}" in body
...@@ -196,3 +196,37 @@ def test_slicing_alone_is_not_enough_and_the_code_says_why(): ...@@ -196,3 +196,37 @@ def test_slicing_alone_is_not_enough_and_the_code_says_why():
# Slicing splits a BATCH; it does nothing for n=1, which is the case that # Slicing splits a BATCH; it does nothing for n=1, which is the case that
# OOMed. Whoever next trims these comments should know that. # OOMed. Whoever next trims these comments should know that.
assert "n=1" in body assert "n=1" in body
# --------------------------------------------------------------------------- #
# an error the reader can act on
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("exc,expect", [
(RuntimeError(), "RuntimeError (raised with no message)"),
(RuntimeError(""), "RuntimeError (raised with no message)"),
(RuntimeError(" "), "RuntimeError (raised with no message)"),
(AssertionError(), "AssertionError (raised with no message)"),
(ValueError("size must be a multiple of 8"),
"ValueError: size must be a multiple of 8"),
])
def test_an_exception_with_no_message_still_says_something(exc, expect):
"""Measured 2026-11-01: Z-Image-Turbo answered "Image generation failed ...
so this is not a load failure." and then stopped, because str(exc) was
empty. The reader could not tell a shape error from a full card."""
assert images_mod._exc_text(exc) == expect
def test_the_type_is_not_repeated_when_the_message_already_carries_it():
"""diffusers and torch both raise messages that start with their own class
name; 'RuntimeError: RuntimeError: ...' reads like a bug in the reporter."""
assert images_mod._exc_text(
RuntimeError("RuntimeError: already named")) == "RuntimeError: already named"
def test_the_generation_path_reports_through_the_helper():
"""Not str(e) interpolated raw -- that is the shape that produced the
message with nothing after it."""
body = SRC[SRC.index("if _loaded_diffusers is not None:"):][:2200]
assert "_exc_text(e)" in body, "the detail has to go through the helper"
assert "else str(e)" not in body, "the raw-str fallback is what broke"
...@@ -10,6 +10,12 @@ Krea 2 and the Qwen-Image family return ``(prompt_embeds, prompt_embeds_mask)`` ...@@ -10,6 +10,12 @@ Krea 2 and the Qwen-Image family return ``(prompt_embeds, prompt_embeds_mask)``
which reads like a missing argument rather than a mislabelled one, and it which reads like a missing argument rather than a mislabelled one, and it
arrives wrapped in "Failed to load image model", minutes into a 7.4B load. arrives wrapped in "Failed to load image model", minutes into a 7.4B load.
MOSS-Image and Kromoss do the same thing and name the parameter
``prompt_attention_mask`` instead, so a check for the single name
``prompt_embeds_mask`` missed them and their boolean mask went over as a
negative. See :mod:`tests.test_encode_prompt_two_tuple`, which covers the
shape-based discriminator that replaced it.
""" """
import inspect import inspect
...@@ -21,11 +27,13 @@ def _source(): ...@@ -21,11 +27,13 @@ def _source():
def test_the_mask_families_are_told_apart_by_signature_not_by_name(): def test_the_mask_families_are_told_apart_by_signature_not_by_name():
"""A name list would have to grow for every new family; the pipelines that """A name list would have to grow for every new family -- which is exactly
need a mask say so in their own __call__ signature.""" what happened: the one name here did not cover MOSS. The signature is still
asked first, for both spellings, and the tensors decide when it is silent."""
src = _source() src = _source()
assert "'prompt_embeds_mask' in _call_params" in src assert "_second_is_a_mask(enc, _call_params)" in src
# And the discriminator comes from the pipeline itself. assert '"prompt_embeds_mask", "prompt_attention_mask"' in src
# And the discriminator still comes from the pipeline itself.
assert "_inspect.signature(pipeline.__call__).parameters" in src assert "_inspect.signature(pipeline.__call__).parameters" in src
...@@ -33,8 +41,8 @@ def test_a_mask_pipeline_falls_back_to_its_own_encoder(): ...@@ -33,8 +41,8 @@ def test_a_mask_pipeline_falls_back_to_its_own_encoder():
"""Weighted 75-token windows are a CLIP-pair feature and mean nothing to """Weighted 75-token windows are a CLIP-pair feature and mean nothing to
these pipelines, so the right answer is to pass the raw prompt.""" these pipelines, so the right answer is to pass the raw prompt."""
src = _source() src = _source()
i = src.index("'prompt_embeds_mask' in _call_params") i = src.index("_second_is_a_mask(enc, _call_params)")
branch = src[i:i + 1200] branch = src[i:i + 1400]
assert "embed_kwargs = {}" in branch assert "embed_kwargs = {}" in branch
# The SD 1.x reading must still exist, just after it. # The SD 1.x reading must still exist, just after it.
assert "elif len(enc) == 2:" in branch assert "elif len(enc) == 2:" in branch
......
# 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.
"""bitsandbytes cannot quantize a single-file component as it loads.
``from_single_file`` given a bnb config takes the PRE-quantized path: it
expects the checkpoint to carry bnb metadata already. A plain community
fine-tune carries none, so the load reads the whole file and dies with
Supplied state dict for transformer_blocks.0.attn.to_gate.weight does not
contain `bitsandbytes__*` and possibly other `quantized_stats` components
-- a tensor name instead of a cause. Measured 2026-11-01 on Krea-2-RealDream
(``civitai:3396368:3304483``, 12.9 GB), whose docstring had promised this
worked. It is refused up front now, naming what does work.
"""
import pytest
from codai.models import hf_loading
SRC = open("codai/models/hf_loading.py").read()
def _cfg(source, quant):
return {"component_models": {"transformer": source},
"component_quantization": quant,
"path": "unsloth/Krea-2-Turbo"}
@pytest.mark.parametrize("source", [
"civitai:3396368:3304483",
"/AI/models/realDream_krea2V3.safetensors",
])
@pytest.mark.parametrize("mode", ["4bit", "8bit"])
def test_bnb_on_a_single_file_component_is_refused(source, mode):
with pytest.raises(RuntimeError) as ei:
hf_loading.build_component_models(
"unsloth/Krea-2-Turbo", _cfg(source, {"transformer": mode}), None)
msg = str(ei.value)
assert "single file" in msg
assert mode in msg
assert "transformer" in msg
def test_the_refusal_names_what_does_work():
"""A refusal that does not say what to do instead just moves the dead end."""
with pytest.raises(RuntimeError) as ei:
hf_loading.build_component_models(
"unsloth/Krea-2-Turbo",
_cfg("civitai:3396368:3304483", {"transformer": "4bit"}), None)
msg = str(ei.value)
assert "GGUF" in msg, "the GGUF route is the one that quantizes on load"
assert "offload" in msg
def test_it_refuses_before_reading_the_file_and_before_the_class_lookup():
"""12.9 GB read and then an error about a tensor name is the behaviour being
replaced, so the check must precede the load. It must also precede the class
lookup: that lookup `continue`s when the installed diffusers has no class
for the component, which skipped the refusal entirely and is how this test
caught it."""
head = SRC.index("def build_component_models")
body = SRC[head:SRC.index("def ", head + 10)]
refusal = body.index("from_single_file cannot do")
lookup = body.index("cls = getattr(diffusers, cls_name, None)")
load = body.index("cls.from_single_file(")
assert refusal < lookup < load
def test_an_unquantized_single_file_component_is_untouched():
"""The ordinary case -- a single-file component with no quantization asked
for -- must not be caught by this."""
cfg = _cfg("civitai:3396368:3304483", {})
# No RuntimeError about single files; it fails later or succeeds, either way
# not here. diffusers may be absent in the test environment, which returns
# ({}, '') -- also acceptable, and not a refusal.
try:
hf_loading.build_component_models("unsloth/Krea-2-Turbo", cfg, None)
except RuntimeError as exc:
assert "from_single_file cannot do" not in str(exc)
def test_a_repo_component_may_still_be_quantized():
"""The capability is real for a repo; only the single-file path lacks it."""
cfg = _cfg("prithivMLmods/Some-Transformer", {"transformer": "4bit"})
try:
hf_loading.build_component_models("unsloth/Krea-2-Turbo", cfg, None)
except RuntimeError as exc:
assert "from_single_file cannot do" not in str(exc)
def test_the_docstring_no_longer_promises_it():
"""The false promise is why the config change was tried at all."""
head = SRC.index("def build_component_models")
doc = SRC[head:SRC.index('"""', SRC.index('"""', head) + 3)]
assert "does NOT apply to a component that is a single FILE" in doc
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment