pgw#1514: skeleton.build asks WHY before saying WHAT — a collected tree is not an absent index

src/gen_worker/serving/streaming/skeleton.py

@@ -28,6 +28,7 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Dict, List, Mapping, Optional, Tuple
 
+from ...models import projection
 from ...models.meta_init import init_empty_weights
 
 if TYPE_CHECKING:  # pragma: no cover - typing only
@@ -153,6 +154,22 @@ def build(
     extra_kwargs: Optional[Mapping[str, Any]] = None,
 ) -> Skeleton:
     """Build ``pipeline_cls`` from configs only. Reads no tensor bytes."""
+    # pgw#1514: ASK WHY BEFORE SAYING WHAT, and ask it ONCE for the whole tree.
+    # `Path.is_file()` FOLLOWS the link, so "this tree never had an index" and
+    # "this tree's objects were collected" arrive at the check below as the
+    # same False — and the second is a fact about the STORE that this refusal
+    # reported as a fact about the CHECKPOINT. Measured on a real 5.6 GB tree
+    # (se#790): pin dropped, one GC, and a tree whose `model_index.json` is
+    # right there gets refused for not having one.
+    #
+    # The index merely DIES FIRST — 14 entries dangle in that state — so this
+    # walks the whole tree rather than guarding one file, which would only
+    # move the wrong message down to the next component's config. Same helper
+    # and same sentence as the eager bridge uses (pgw#1513), because two
+    # hand-written strings is how this shape reached four callers.
+    collected = projection.collected_entries(checkpoint_dir)
+    if collected:
+        raise SkeletonError(projection.collected_refusal(checkpoint_dir, collected))
     index_path = Path(checkpoint_dir) / MODEL_INDEX
     if not index_path.is_file():
         raise SkeletonError(

tests/test_projected_tree_reading.py

@@ -284,3 +284,96 @@ def test_collected_entries_puts_model_index_json_FIRST(tmp_path: Path) -> None:
     assert got[1:] == sorted(got[1:]), f"the tail must stay sorted: {got}"
     assert "model_index.json" in projection.collected_refusal(tmp_path, got)[:400], (
         "and it must survive into the truncated (shown) window of the message")
+
+
+# ---------------------------------------------------------------------------
+# pgw#1514: the STREAMING reader. Everything above is the eager bridge; this is
+# the other caller, and it reads the index BEFORE anything that guard can see.
+# ---------------------------------------------------------------------------
+
+
+class _KeepsComponents:
+    """A pipeline class that would build happily — so a refusal below is the
+    guard's, never an accident of the fixture."""
+
+    def __init__(self, **components: Any) -> None:  # pragma: no cover
+        for name, value in components.items():
+            setattr(self, name, value)
+
+
+def _collected_tree(root: Path, *, entries: tuple[str, ...]) -> Path:
+    """A projected tree whose objects have been collected: every entry is a
+    dangling link into `objects/`, exactly as one GC pass after a lost pin
+    leaves it."""
+    base = root / "cas"
+    (base / "refs").mkdir(parents=True)
+    (base / "objects").mkdir(parents=True)
+    tree = base / "snapshots" / ("sha256:" + "c0" * 32)
+    tree.mkdir(parents=True)
+    for rel in entries:
+        path = tree / rel
+        path.parent.mkdir(parents=True, exist_ok=True)
+        depth = len(Path(rel).parts)
+        up = Path(*([".."] * (depth + 1)))
+        path.symlink_to(up / "objects" / "sha256" / "de" / "ad" / ("de" * 32))
+        assert path.is_symlink() and not path.exists(), rel
+    return tree
+
+
+def test_skeleton_build_does_not_call_a_collected_tree_index_less(
+    tmp_path: Path,
+) -> None:
+    """THE pgw#1514 REGRESSION. `Path.is_file()` follows the link, so before
+    this fix a dangling `model_index.json` and an absent one were the same
+    False — and `skeleton.build` reported the store's condition as the
+    checkpoint's shape."""
+    from gen_worker.serving.streaming import skeleton as sk
+
+    tree = _collected_tree(
+        tmp_path, entries=("model_index.json", "dit/config.json", "vae/config.json")
+    )
+    with pytest.raises(sk.SkeletonError) as caught:
+        sk.build(_KeepsComponents, tree)
+
+    message = str(caught.value)
+    assert "COLLECTED" in message, message
+    assert "RE-FETCHED" in message, message
+    assert not message.startswith(f"{tree} carries no model_index.json"), (
+        "the false refusal is exactly what this issue removed")
+
+
+def test_skeleton_build_walks_the_WHOLE_tree_not_just_the_index(
+    tmp_path: Path,
+) -> None:
+    """The index merely DIES FIRST. 14 entries dangle in the measured state, so
+    guarding one file would relocate the wrong message to the next component's
+    config rather than remove it. Here the index is FINE and a component config
+    is collected — the pre-fix code reached `_build_on_meta` and failed there
+    with a message about a missing config."""
+    from gen_worker.serving.streaming import skeleton as sk
+
+    tree = _collected_tree(tmp_path, entries=("dit/config.json",))
+    (tree / "model_index.json").write_text(
+        '{"_class_name": "X", "dit": ["anima.components", "AnimaDiTComponent"]}'
+    )
+
+    with pytest.raises(sk.SkeletonError) as caught:
+        sk.build(_KeepsComponents, tree)
+    assert "COLLECTED" in str(caught.value), str(caught.value)
+    assert "dit/config.json" in str(caught.value)
+
+
+def test_a_tree_that_GENUINELY_has_no_index_still_says_so(tmp_path: Path) -> None:
+    """The original wording SURVIVES and is now true whenever it is reached. A
+    fix that swallowed the genuine case would trade one wrong message for
+    another, which is the mistake this issue is about."""
+    from gen_worker.serving.streaming import skeleton as sk
+
+    bare = tmp_path / "bare"
+    (bare / "unet").mkdir(parents=True)
+    (bare / "unet" / "config.json").write_text("{}")
+
+    with pytest.raises(sk.SkeletonError) as caught:
+        sk.build(_KeepsComponents, bare)
+    assert "carries no model_index.json" in str(caught.value)
+    assert "COLLECTED" not in str(caught.value)