pipeline-cache: reject corrupt-JSON caches (stop the empty-tokenizer death spiral)

A cache dir's completion marker only proves the save FINISHED, not that every
file landed intact. A transient truncated write — repeatedly a 0-byte
tokenizer/tokenizer_config.json — slipped into an otherwise "valid" cache and
then threw JSONDecodeError on EVERY subsequent video load, knocking the pipeline
off its fast path into the offload fallback ladder (balanced→sequential→disk),
which churned for hours, leaked ~22 GB VRAM, and died on a meta-tensor error.

Add _first_bad_json(dir): walk the (small) JSONs and flag the first that is
empty or unparseable — the big weights are .safetensors and aren't scanned, so
it's cheap. Wire it in on BOTH sides:
 - load: valid() and component_valid() now invalidate + return False when any
   cached JSON is corrupt, so a poisoned cache becomes a clean rebuild instead
   of a death spiral.
 - save: save()/save_component() verify the temp dir before committing, and
   mark_monolithic_complete() refuses to finalize a dir with a corrupt JSON —
   so a truncated write is never cached in the first place.
Added invalidate_path(p) helper.

Verified: _first_bad_json flags 0-byte and garbage JSONs, passes clean ones.
The already-poisoned Wan2.2-VACE cache was deleted out-of-band to unblock.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RdMufYvtTbtGDWsiZVoXce

codai/models/pipeline_cache.py

@@ -106,6 +106,35 @@ def _marker(p: str) -> str:
     return os.path.join(p, ".coderai_pipeline_cache.json")
 
 
+def _first_bad_json(d: str) -> Optional[str]:
+    """Return the path of the first empty/unparseable ``.json`` under ``d``, or None
+    if every JSON is intact.
+
+    The completion markers only prove a save *finished*, not that every file landed
+    intact — a transient truncated write (e.g. a 0-byte ``tokenizer_config.json``)
+    slips into an otherwise "valid" cache and then throws JSONDecodeError on EVERY
+    subsequent load, knocking the video pipeline off its fast path into the offload
+    fallback ladder (and a VRAM leak). Checking the small JSONs is cheap (they're
+    KBs; the big weights are ``.safetensors``, not scanned) and turns a poisoned
+    cache into a clean rebuild instead of a death spiral."""
+    try:
+        for root, _, files in os.walk(d):
+            for fn in files:
+                if not fn.endswith(".json"):
+                    continue
+                fp = os.path.join(root, fn)
+                try:
+                    if os.path.getsize(fp) == 0:
+                        return fp
+                    with open(fp) as f:
+                        json.load(f)
+                except Exception:
+                    return fp
+    except Exception:
+        return None
+    return None
+
+
 def valid(p: str) -> bool:
     """True if a complete, current cache exists at ``p`` and rebuild wasn't forced."""
     if not p or _force_rebuild():
@@ -115,7 +144,15 @@ def valid(p: str) -> bool:
             return False
         with open(_marker(p)) as f:
             meta = json.load(f)
-        return meta.get("version") == _CACHE_VERSION and meta.get("complete") is True
+        if not (meta.get("version") == _CACHE_VERSION and meta.get("complete") is True):
+            return False
+        bad = _first_bad_json(p)
+        if bad:
+            print(f"  [pipeline-cache] cache {os.path.basename(p)} has a corrupt JSON "
+                  f"({os.path.relpath(bad, p)}) — invalidating so it rebuilds clean")
+            invalidate_path(p)
+            return False
+        return True
     except Exception:
         return False
 
@@ -144,6 +181,11 @@ def mark_monolithic_complete(model_name: str, model_cfg: Optional[dict]) -> bool
         p = path(model_name, model_cfg)
         if not os.path.isfile(os.path.join(p, "model_index.json")):
             return False
+        _bad = _first_bad_json(p)
+        if _bad:
+            print(f"  [pipeline-cache] NOT finalizing {os.path.basename(p)}: corrupt "
+                  f"JSON ({os.path.relpath(_bad, p)}) — leaving incremental, will rebuild")
+            return False
         with open(_marker(p), "w") as f:
             json.dump({
                 "version": _CACHE_VERSION, "complete": True,
@@ -175,9 +217,17 @@ def component_valid(model_name: str, model_cfg: Optional[dict], comp: str) -> bo
             return False
         with open(_component_marker(cdir)) as f:
             meta = json.load(f)
-        return (meta.get("version") == _CACHE_VERSION
+        if not (meta.get("version") == _CACHE_VERSION
                 and meta.get("complete") is True
-                and meta.get("signature") == _signature(model_name, model_cfg))
+                and meta.get("signature") == _signature(model_name, model_cfg)):
+            return False
+        bad = _first_bad_json(cdir)
+        if bad:
+            print(f"  [pipeline-cache] component '{comp}' has a corrupt JSON "
+                  f"({os.path.basename(bad)}) — invalidating so it rebuilds clean")
+            invalidate_path(cdir)
+            return False
+        return True
     except Exception:
         return False
 
@@ -201,6 +251,9 @@ def save_component(comp_obj, model_name: str, model_cfg: Optional[dict], comp: s
         print(f"  [pipeline-cache] caching component '{comp}' → {cdir}")
         t0 = time.time()
         comp_obj.save_pretrained(tmp)
+        _bad = _first_bad_json(tmp)
+        if _bad:
+            raise ValueError(f"save produced a corrupt JSON ({os.path.basename(_bad)})")
         with open(_component_marker(tmp), "w") as f:
             json.dump({
                 "version": _CACHE_VERSION, "complete": True,
@@ -236,11 +289,10 @@ def _dir_size(d: str) -> int:
     return total
 
 
-def invalidate(model_name: str, model_cfg: Optional[dict]) -> None:
-    """Delete a model's cache dir (e.g. after a failed cache load) so the next
-    build rewrites it. Best-effort."""
+def invalidate_path(p: str) -> None:
+    """Delete a cache dir by path (best-effort). Used when a poisoned/corrupt cache
+    is detected during validation."""
     try:
-        p = path(model_name, model_cfg)
         if p and os.path.isdir(p):
             shutil.rmtree(p, ignore_errors=True)
             print(f"  [pipeline-cache] invalidated {p}")
@@ -248,6 +300,15 @@ def invalidate(model_name: str, model_cfg: Optional[dict]) -> None:
         pass
 
 
+def invalidate(model_name: str, model_cfg: Optional[dict]) -> None:
+    """Delete a model's cache dir (e.g. after a failed cache load) so the next
+    build rewrites it. Best-effort."""
+    try:
+        invalidate_path(path(model_name, model_cfg))
+    except Exception:
+        pass
+
+
 def sweep_stale(max_age_s: float = 1800.0) -> None:
     """Delete orphaned ``*.building`` temp dirs left by a save that was killed
     mid-write (the atomic ``os.replace`` never ran). They waste tens of GB and,
@@ -313,6 +374,9 @@ def save(pipe, p: str, *, model_name: str = "", model_cfg: Optional[dict] = None
         print(f"  [pipeline-cache] saving quantized pipeline → {p}")
         t0 = time.time()
         pipe.save_pretrained(tmp)
+        _bad = _first_bad_json(tmp)
+        if _bad:
+            raise ValueError(f"save produced a corrupt JSON ({os.path.relpath(_bad, tmp)})")
         with open(_marker(tmp), "w") as f:
             json.dump({
                 "version": _CACHE_VERSION, "complete": True,