longcat: serve the avatar families, which ship no text encoder (v0.2.72)

LongCat-Video-Avatar-1.5 is a transformer and an audio encoder: base_model/,
base_model_int8/, lora/dmd_lora, scheduler/, whisper-large-v3/. It has NO
tokenizer, text_encoder or vae. Its README says so in its own frontmatter
(`base_model: meituan-longcat/LongCat-Video`) and upstream's instructions
download both repos side by side.

Validated against the base checkpoint's layout it reported four missing
directories that were never part of the repo, and the loader would then have
asked from_pretrained for a text_encoder/ that does not exist — and for
subfolder="dit", where the avatar transformer is base_model/.

So the layout is family-aware now. An avatar checkpoint contributes its
scheduler, its transformer (base_model/ at bf16, base_model_int8/ at int8) and
its LoRA; tokenizer/text_encoder/vae are borrowed from a base checkpoint, named
by `base_model` per model or falling back to the upstream repo id, which
resolve_checkpoint() turns into a local snapshot. A base checkpoint that is
absent or incomplete is reported AS the base checkpoint, rather than as a
tokenizer missing from the avatar repo, which sends you looking in the wrong
place.

Either transformer alone is enough to validate: base_model/ is 31.7 GB of bf16
that cannot fit a 24 GB card, so it is deliberately not downloaded here.

Verified against the real checkpoint: family avatar-1.5, shared components
resolving to the base snapshot, 4 INT8 shards with 2241 keys matching the index
and 633 quantised Linear layers (the base DiT has 437 — the difference is the
audio cross-attention), dmd_lora present. All three validators return [].
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0185zEBiy3KWB37xNGSrxL7w

codai/__init__.py

@@ -16,7 +16,7 @@
 
 # Canonical product version for CoderAI — single source of truth. Both the API
 # metadata and the admin web UI read from here.
-__version__ = "0.2.71"
+__version__ = "0.2.72"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/api/longcat_worker.py

@@ -45,6 +45,7 @@ Per-model config keys (models.json entry, all optional)::
     "quality": "fast",             # draft | fast | best
     "offload_strategy": "",        # '' | model | sequential
     "bsa": "off",                  # block-sparse attention: on | off | auto
+    "base_model": "",              # avatar families borrow tokenizer/text_encoder/vae
     "cp_split_hw": "",             # context-parallel tile, e.g. "1x2"
     "used_vram_gb": 0,             # what to reserve before starting; 0 = measured
     "gpu_device": 0
@@ -494,6 +495,11 @@ def ensure_service(model_path: str, config: dict = None,
         split = str(config.get("cp_split_hw") or "").strip()
         if split:
             cmd += ["--cp-split-hw", split]
+        # An avatar checkpoint has no tokenizer/text_encoder/vae of its own; blank
+        # lets the service fall back to the upstream base repo.
+        base = str(config.get("base_model") or "").strip()
+        if base:
+            cmd += ["--base-model", base]
         if sec is not None and str(getattr(sec, "extra_args", "") or "").strip():
             import shlex
             cmd += shlex.split(str(sec.extra_args))

tests/test_longcat_avatar_layout.py

+"""An avatar checkpoint is a transformer and an audio encoder, nothing else.
+
+LongCat-Video-Avatar-1.5 ships base_model/, base_model_int8/, lora/dmd_lora, scheduler/
+and whisper-large-v3/ — and NO tokenizer, text_encoder or vae. Its README says so in
+its frontmatter (`base_model: meituan-longcat/LongCat-Video`) and upstream's own
+instructions download both repos. Validating it against the base checkpoint's layout
+reports four missing directories that were never supposed to be there.
+"""
+import importlib.util
+import pathlib
+
+import pytest
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+COMMON = ROOT / "tools" / "longcat_common.py"
+SERVICE = ROOT / "tools" / "longcat_service.py"
+
+
+@pytest.fixture
+def LC():
+    spec = importlib.util.spec_from_file_location("lc_avatar_layout", COMMON)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+def _base_checkpoint(path):
+    for sub in ("tokenizer", "text_encoder", "vae", "scheduler", "dit"):
+        (path / sub).mkdir(parents=True)
+    return path
+
+
+def _avatar_checkpoint(path, int8=True, bf16=True):
+    (path / "scheduler").mkdir(parents=True)
+    if bf16:
+        (path / "base_model").mkdir()
+    if int8:
+        (path / "base_model_int8").mkdir()
+    (path / "lora").mkdir()
+    (path / "lora" / "dmd_lora.safetensors").write_bytes(b"")
+    return path
+
+
+# ---------------------------------------------------------------- family detection
+
+@pytest.mark.parametrize("ref,want", [
+    ("meituan-longcat/LongCat-Video", ""),
+    ("meituan-longcat/LongCat-Video-Avatar", "avatar"),
+    ("meituan-longcat/LongCat-Video-Avatar-1.5", "avatar-1.5"),
+    ("/AI/x/models--meituan-longcat--LongCat-Video-Avatar-1.5/snapshots/a", "avatar-1.5"),
+])
+def test_the_family_comes_from_the_path(LC, ref, want):
+    assert LC.family_of(ref) == want
+
+
+# ---------------------------------------------------------------- the dit subfolder
+
+def test_the_avatar_transformer_is_under_base_model(LC):
+    """The base checkpoint uses dit/; loading an avatar with subfolder="dit" fails."""
+    assert LC.dit_subdir("avatar-1.5") == "base_model"
+    assert LC.dit_subdir("avatar") == "base_model"
+    assert LC.dit_subdir("") == "dit"
+
+
+# ---------------------------------------------------------------- shared components
+
+def test_the_base_checkpoint_shares_with_itself(LC, tmp_path):
+    base = _base_checkpoint(tmp_path / "base")
+    assert LC.shared_checkpoint(str(base)) == str(base)
+
+
+def test_an_avatar_borrows_from_the_configured_checkpoint(LC, tmp_path):
+    base = _base_checkpoint(tmp_path / "base")
+    avatar = _avatar_checkpoint(tmp_path / "LongCat-Video-Avatar-1.5")
+    assert LC.shared_checkpoint(str(avatar), str(base)) == str(base)
+
+
+def test_an_avatar_with_nothing_configured_looks_for_the_upstream_base(LC, tmp_path,
+                                                                      monkeypatch):
+    """Unresolvable comes back as the repo id, which the validator then names."""
+    monkeypatch.setenv("HF_HOME", str(tmp_path / "empty"))
+    # resolve_checkpoint also falls back to ~/.cache/huggingface/hub, which on this
+    # machine is the real store — point HOME somewhere empty so the test is about the
+    # fallback and not about what happens to be downloaded.
+    monkeypatch.setenv("HOME", str(tmp_path / "home"))
+    avatar = _avatar_checkpoint(tmp_path / "LongCat-Video-Avatar-1.5")
+    assert LC.shared_checkpoint(str(avatar)) == LC.BASE_REPO
+
+
+# ---------------------------------------------------------------- validation
+
+def test_a_valid_avatar_checkpoint_passes_with_a_base(LC, tmp_path):
+    base = _base_checkpoint(tmp_path / "base")
+    avatar = _avatar_checkpoint(tmp_path / "LongCat-Video-Avatar-1.5")
+    assert LC.checkpoint_problems(str(avatar), (), "avatar-1.5", str(base)) == []
+
+
+def test_the_avatar_is_not_asked_for_components_it_never_ships(LC, tmp_path):
+    """The bug this fixes: four 'missing' directories that are not part of the repo."""
+    base = _base_checkpoint(tmp_path / "base")
+    avatar = _avatar_checkpoint(tmp_path / "LongCat-Video-Avatar-1.5")
+    problems = " ".join(LC.checkpoint_problems(str(avatar), (), "avatar-1.5", str(base)))
+    for never_there in ("tokenizer/", "text_encoder/", "vae/", "dit/"):
+        assert f"missing {never_there} in {avatar}" not in problems
+
+
+def test_an_avatar_without_any_transformer_is_caught(LC, tmp_path):
+    base = _base_checkpoint(tmp_path / "base")
+    avatar = _avatar_checkpoint(tmp_path / "LongCat-Video-Avatar-1.5",
+                                int8=False, bf16=False)
+    problems = LC.checkpoint_problems(str(avatar), (), "avatar-1.5", str(base))
+    assert any("needs at least one transformer" in p for p in problems)
+
+
+@pytest.mark.parametrize("int8,bf16", [(True, False), (False, True)])
+def test_either_transformer_alone_is_enough(LC, tmp_path, int8, bf16):
+    """We deliberately skip base_model/ on a 24 GB card — int8 alone must validate."""
+    base = _base_checkpoint(tmp_path / "base")
+    avatar = _avatar_checkpoint(tmp_path / "LongCat-Video-Avatar-1.5",
+                                int8=int8, bf16=bf16)
+    assert LC.checkpoint_problems(str(avatar), (), "avatar-1.5", str(base)) == []
+
+
+def test_a_missing_base_checkpoint_is_reported_as_such(LC, tmp_path):
+    """Not as "missing tokenizer/" inside the avatar, which sends you looking in the
+    wrong repo."""
+    avatar = _avatar_checkpoint(tmp_path / "LongCat-Video-Avatar-1.5")
+    problems = LC.checkpoint_problems(str(avatar), (), "avatar-1.5",
+                                      str(tmp_path / "nope"))
+    assert any("base checkpoint" in p for p in problems)
+
+
+def test_an_incomplete_base_checkpoint_names_the_base(LC, tmp_path):
+    base = tmp_path / "base"
+    (base / "tokenizer").mkdir(parents=True)      # vae and text_encoder missing
+    avatar = _avatar_checkpoint(tmp_path / "LongCat-Video-Avatar-1.5")
+    problems = LC.checkpoint_problems(str(avatar), (), "avatar-1.5", str(base))
+    assert any("text_encoder/ in the base checkpoint" in p for p in problems)
+    assert any("vae/ in the base checkpoint" in p for p in problems)
+
+
+def test_the_base_checkpoint_still_validates_the_old_way(LC, tmp_path):
+    base = _base_checkpoint(tmp_path / "base")
+    assert LC.checkpoint_problems(str(base), (), "", str(base)) == []
+    (base / "vae").rmdir()
+    assert any("missing vae/" in p for p in LC.checkpoint_problems(str(base), (), ""))
+
+
+# ---------------------------------------------------------------- service wiring
+
+def test_the_service_loads_shared_components_from_the_shared_root():
+    src = SERVICE.read_text(encoding="utf-8")
+    body = src[src.index("def load_pipeline"):src.index("def _cp_split_hw")]
+    for component in ("tokenizer", "text_encoder", "vae"):
+        assert f'shared, subfolder="{component}"' in body, component
+    # the scheduler is the avatar's OWN
+    assert 'root, subfolder="scheduler"' in body
+    assert "LC.dit_subdir(family)" in body

tools/longcat_common.py

@@ -30,6 +30,19 @@ import os
 # from_pretrained call takes `checkpoint_dir` plus one of these subfolders.
 CHECKPOINT_SUBDIRS = ("tokenizer", "text_encoder", "vae", "scheduler", "dit")
 
+# The avatar checkpoints are a DiT and an audio encoder, nothing else: no tokenizer,
+# no text_encoder, no vae. Their README says so in its frontmatter
+# (`base_model: meituan-longcat/LongCat-Video`) and upstream's own instructions
+# download BOTH repos. So an avatar family borrows the shared components from the base
+# checkpoint and brings only its own transformer, scheduler and LoRA.
+SHARED_SUBDIRS = ("tokenizer", "text_encoder", "vae")
+AVATAR_OWN_SUBDIRS = ("scheduler",)
+# Their bf16 transformer is under base_model/, where the base checkpoint uses dit/.
+AVATAR_DIT_SUBDIR = "base_model"
+BASE_DIT_SUBDIR = "dit"
+# Where the shared components come from when none is configured.
+BASE_REPO = "meituan-longcat/LongCat-Video"
+
 # The two LoRAs that make stages 2 and 3 what they are, under `lora/`.
 LORA_FILES = {
     "distill": "lora/cfg_step_lora.safetensors",
@@ -266,7 +279,27 @@ def resolve_checkpoint(ref: str) -> str:
     return path
 
 
-def checkpoint_problems(checkpoint_dir: str, stages=()) -> list:
+def dit_subdir(family: str = "") -> str:
+    """Which subfolder holds the bf16 transformer for this family."""
+    return AVATAR_DIT_SUBDIR if family else BASE_DIT_SUBDIR
+
+
+def shared_checkpoint(checkpoint_dir: str = "", configured: str = "") -> str:
+    """Where tokenizer/text_encoder/vae come from.
+
+    For the base checkpoint that is itself. An avatar checkpoint does not ship them,
+    so it borrows them from the base weights — configured explicitly, or the upstream
+    repo id, which resolve_checkpoint() turns into a local snapshot.
+    """
+    if configured:
+        return resolve_checkpoint(configured)
+    if family_of(checkpoint_dir):
+        return resolve_checkpoint(BASE_REPO)
+    return checkpoint_dir
+
+
+def checkpoint_problems(checkpoint_dir: str, stages=(), family: str = "",
+                        shared_dir: str = "") -> list:
     """Human-readable reasons this directory cannot serve ``stages``, or [].
 
     Checked BEFORE loading: a missing subfolder otherwise surfaces minutes in, as a
@@ -278,9 +311,31 @@ def checkpoint_problems(checkpoint_dir: str, stages=()) -> list:
     root = os.path.expanduser(checkpoint_dir)
     if not os.path.isdir(root):
         return [f"checkpoint directory does not exist: {root}"]
-    for sub in CHECKPOINT_SUBDIRS:
-        if not os.path.isdir(os.path.join(root, sub)):
-            problems.append(f"missing {sub}/ in {root}")
+    if family:
+        # Its own parts…
+        for sub in AVATAR_OWN_SUBDIRS:
+            if not os.path.isdir(os.path.join(root, sub)):
+                problems.append(f"missing {sub}/ in {root}")
+        if not any(os.path.isdir(os.path.join(root, d))
+                   for d in (AVATAR_DIT_SUBDIR, INT8_SUBDIR)):
+            problems.append(
+                f"missing {AVATAR_DIT_SUBDIR}/ and {INT8_SUBDIR}/ in {root} — an "
+                f"avatar checkpoint needs at least one transformer")
+        # …and the shared ones, from wherever they were borrowed.
+        shared = os.path.expanduser(shared_dir or "")
+        if not shared or not os.path.isdir(shared):
+            problems.append(
+                f"the {family} checkpoint has no tokenizer/text_encoder/vae of its "
+                f"own and no base checkpoint was found to borrow them from "
+                f"(looked for {BASE_REPO})")
+        else:
+            for sub in SHARED_SUBDIRS:
+                if not os.path.isdir(os.path.join(shared, sub)):
+                    problems.append(f"missing {sub}/ in the base checkpoint {shared}")
+    else:
+        for sub in CHECKPOINT_SUBDIRS:
+            if not os.path.isdir(os.path.join(root, sub)):
+                problems.append(f"missing {sub}/ in {root}")
     for stage in stages:
         if stage == "distill":
             rel = LORA_FILES["distill"]

tools/longcat_service.py

@@ -59,7 +59,7 @@ _lock = threading.RLock()
 _state = {
     "pipe": None, "model": "", "source": "", "dtype": "bfloat16",
     "family": "", "variant": "bf16", "use_int8": False, "use_distill": False,
-    "cp_split_hw": [1, 1], "audio_encoder": None,
+    "cp_split_hw": [1, 1], "audio_encoder": None, "base_model": "",
 }
 _progress = {"active": False, "stage": "", "segment": 0, "segments": 0,
              "step": 0, "steps": 0}
@@ -81,7 +81,10 @@ def load_pipeline(checkpoint_dir: str, source_dir: str, dtype: str = "bfloat16",
         if _state["pipe"] is not None:
             return _state["pipe"]
 
-        problems = LC.source_problems(source_dir) + LC.checkpoint_problems(checkpoint_dir)
+        _f = LC.family_of(checkpoint_dir)
+        problems = LC.source_problems(source_dir) + LC.checkpoint_problems(
+            checkpoint_dir, (), _f,
+            LC.shared_checkpoint(checkpoint_dir, _state.get("base_model") or ""))
         if problems:
             raise ValueError("; ".join(problems))
 
@@ -100,15 +103,22 @@ def load_pipeline(checkpoint_dir: str, source_dir: str, dtype: str = "bfloat16",
         td = {"bfloat16": torch.bfloat16, "float16": torch.float16,
               "float32": torch.float32}.get(str(dtype).lower(), torch.bfloat16)
         root = os.path.expanduser(checkpoint_dir)
+        family = LC.family_of(root)
+        # An avatar checkpoint ships a transformer and an audio encoder and nothing
+        # else — no tokenizer, no text_encoder, no vae. Upstream downloads the base
+        # repo alongside it and so do we: the shared components come from there, the
+        # scheduler and transformer from the avatar weights.
+        shared = LC.shared_checkpoint(root, _state.get("base_model") or "")
         log(f"loading from {root} (dtype={dtype}, offload={offload or 'none'}) …")
+        if family and shared != root:
+            log(f"family '{family}': tokenizer/text_encoder/vae from {shared}")
 
-        tokenizer = AutoTokenizer.from_pretrained(root, subfolder="tokenizer")
+        tokenizer = AutoTokenizer.from_pretrained(shared, subfolder="tokenizer")
         text_encoder = UMT5EncoderModel.from_pretrained(
-            root, subfolder="text_encoder", torch_dtype=td)
-        vae = AutoencoderKLWan.from_pretrained(root, subfolder="vae", torch_dtype=td)
+            shared, subfolder="text_encoder", torch_dtype=td)
+        vae = AutoencoderKLWan.from_pretrained(shared, subfolder="vae", torch_dtype=td)
         scheduler = FlowMatchEulerDiscreteScheduler.from_pretrained(
             root, subfolder="scheduler")
-        family = LC.family_of(root)
         use_int8 = bool(_state.get("use_int8"))
         use_distill_lora = bool(_state.get("use_distill"))
 
@@ -143,7 +153,7 @@ def load_pipeline(checkpoint_dir: str, source_dir: str, dtype: str = "bfloat16",
                     log("WARNING: this torch has no float8_e4m3fn — loading in "
                         f"{dtype} instead")
             dit = LongCatVideoTransformer3DModel.from_pretrained(
-                root, subfolder="dit", **kw, **_dit_kwargs())
+                root, subfolder=LC.dit_subdir(family), **kw, **_dit_kwargs())
 
         if family:
             # The avatar families are a DIFFERENT pipeline class with its own methods
@@ -782,7 +792,10 @@ def generate(body: dict) -> dict:
 
     stages = LC.resolve_stages(body.get("quality") or "fast", body.get("stage") or "")
     checkpoint = LC.resolve_checkpoint(body.get("model") or _state["model"])
-    problems = LC.checkpoint_problems(checkpoint, stages)
+    _family = LC.family_of(checkpoint)
+    _shared = LC.shared_checkpoint(checkpoint, body.get("base_model")
+                                   or _state.get("base_model") or "")
+    problems = LC.checkpoint_problems(checkpoint, stages, _family, _shared)
     if problems:
         raise ValueError("; ".join(problems))
 
@@ -813,6 +826,8 @@ def generate(body: dict) -> dict:
     if problems:
         raise ValueError("; ".join(problems))
 
+    if body.get("base_model"):
+        _state["base_model"] = str(body["base_model"])
     _state.update(variant=variant, use_int8=use_int8, use_distill=use_distill_lora)
     pipe = load_pipeline(checkpoint, body.get("source") or _state["source"],
                          body.get("dtype") or _state["dtype"],
@@ -1073,6 +1088,9 @@ def main(argv=None):
     ap.add_argument("--offload", default="", help="'' | model | sequential")
     ap.add_argument("--attention", default="xformers",
                     help="xformers | flash | flash3 (falls back to what is installed)")
+    ap.add_argument("--base-model", default="",
+                    help="checkpoint holding tokenizer/text_encoder/vae for an avatar "
+                         "family; blank uses the upstream base repo")
     ap.add_argument("--bsa", default="off",
                     help="block-sparse attention: on | off | auto (auto stays off "
                          "when triton is missing instead of failing)")
@@ -1094,6 +1112,7 @@ def main(argv=None):
     # by default (requirements-longcat.txt installs xformers instead).
     os.environ.setdefault("LONGCAT_ATTENTION", args.attention)
     os.environ.setdefault("LONGCAT_BSA", args.bsa)
+    _state["base_model"] = args.base_model or ""
     # Validate both before anything slow happens: a bad value should not surface after
     # a multi-minute load.
     _bsa_enabled()