release 0.45.0: pgw#622 eager-while-compiling with hot-swap

CHANGELOG.md

@@ -1,5 +1,23 @@
 # Changelog
 
+## 0.45.0 (2026-07-22)
+
+- **pgw#622: eager-while-compiling with hot-swap — a novel request shape
+  serves immediately.** A compiled target's first call at an unseen input
+  signature no longer stalls 30-60s behind Dynamo+Inductor: the consumer
+  guard serves the request (and followups at that signature) through the
+  EAGER original, one background thread warms the compiled callable with a
+  zero-filled dummy batch of the same signature (separate CUDA stream,
+  thread nice +10), and a successful warm atomically hot-swaps the
+  signature onto the compiled path. Each completed warm repacks the live
+  inductor/triton cache root and republishes the cell under the same key
+  (mode=replace) so no other worker ever compiles that (shape, GPU, lane)
+  again. Sequential compile-then-serve is preserved for: the boot
+  warmup-proof window, mandatory quantized lanes (w8a8/w4a4 — eager is not
+  a production lane there), tight VRAM headroom (degrade, never OOM),
+  regional mode, and signature-vocabulary explosions (per-request scalar
+  leaking into signatures disables concurrent routing loudly).
+
 ## 0.44.0 (2026-07-21)
 
 - **pgw#617: hierarchical slot bindings (th#980 companion).**

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "gen-worker"
-version = "0.44.0"
+version = "0.45.0"
 description = "A library used to build custom functions in Cozy Creator's serverless function platform."
 readme = "README.md"
 license = "MIT"

src/gen_worker/compile_cache.py

@@ -1508,10 +1508,20 @@ def wrapper(*args: Any, **kwargs: Any) -> Any:
             if fail_closed:
                 raise CompiledLaneUnavailableError(state["detail"])
             return original(*args, **kwargs)
+        # pgw#622: a novel input signature serves EAGER immediately while a
+        # background thread warms the compiled path, then hot-swaps.
+        router = (failure_signal or {}).get("router")
+        sig = None
+        if router is not None:
+            verdict, sig = router.route(label, compiled, args, kwargs)
+            if verdict == "eager":
+                return original(*args, **kwargs)
         before = proof_before()
         try:
             result = compiled(*args, **kwargs)
             record_success(before)
+            if router is not None:
+                router.mark_warm(sig)
             return result
         except Exception as exc:  # noqa: BLE001 — optional lanes may use eager
             state["failed"] = True
@@ -1708,12 +1718,17 @@ def apply(
     from .models.loading import pipeline_weight_lane
 
     fail_closed = pipeline_weight_lane(pipeline).startswith(("w8a8", "w4a4"))
+    from . import hot_swap
+
     failure_signal: Dict[str, Any] = {
         "callback": None,
         "lock": threading.Lock(),
         "successful_calls": 0,
         "cache_hits": 0,
         "cache_misses": 0,
+        # pgw#622: whole-graph consumer guards route novel signatures
+        # through this; sequential until hot_swap.enable() post-proof.
+        "router": hot_swap.Router(fail_closed=fail_closed) if guard else None,
     }
     applied: list[str] = []
     originals: list[Tuple[Any, str, Callable[..., Any]]] = []
@@ -1833,6 +1848,11 @@ def unwrap(pipeline: Any) -> bool:
     marker = getattr(pipeline, _MARKER_ATTR, None)
     if marker is None:
         return False
+    signal = marker.get("failure_signal")
+    if isinstance(signal, dict):
+        router = signal.get("router")
+        if router is not None:
+            router.close()  # pgw#622: in-flight background warms discard
     for owner, attr, fn in marker.get("originals") or ():
         try:
             setattr(owner, attr, fn)

src/gen_worker/executor.py

@@ -2749,6 +2749,20 @@ def _install_compile_targets(
                     "compile target %s has no runtime guard revocation signal; "
                     "advertising eager", incarnation_id,
                 )
+            if target.active_compile_ref:
+                # pgw#622: post-proof, novel request shapes serve eager while
+                # the compiled path warms in the background; each completed
+                # warm republishes the grown cell for the fleet.
+                from . import hot_swap
+
+                if hot_swap.enable(
+                    pipeline,
+                    on_warmed=hot_swap.Debounce(
+                        self._shape_warm_republisher(spec, pipeline)),
+                ):
+                    logger.info(
+                        "hot-swap: eager-while-compiling enabled for %s",
+                        spec.name)
         if requested_lane and not rec.compile_targets:
             raise compile_cache.CompiledLaneUnavailableError(
                 f"{requested_lane.upper()} setup for {spec.name!r} produced "
@@ -5569,6 +5583,27 @@ def _cell_publisher(self) -> "fleet_cells.CellPublisher":
                 getattr(self._settings, "worker_image_digest", "") or ""),
         )
 
+    def _shape_warm_republisher(
+        self, spec: EndpointSpec, pipeline: Any,
+    ) -> Callable[[], None]:
+        """Republish the grown cell after a background novel-shape warm
+        (pgw#622). Runs on the Debounce thread, never the serving path."""
+        cfg = spec.compile
+        family = str(getattr(cfg, "family", "") or "")
+
+        def republish() -> None:
+            from . import fleet_cells
+
+            cache_dir = self.store._cache_dir
+            live_root = (
+                Path(cache_dir) if cache_dir
+                else Path.home() / ".cache" / "gen-worker"
+            ) / "compile-cache"
+            fleet_cells.republish_after_shape_warm(
+                pipeline, cfg, family, self._cell_publisher(), live_root)
+
+        return republish
+
     def _enable_compiled(
         self, pipe: Any, cfg: Any, artifact: Optional[Path],
     ) -> "fleet_cells.ArmOutcome":

src/gen_worker/fleet_cells.py

@@ -584,6 +584,66 @@ def withhold_self_mint_publish(pending: "PendingSelfMint", reason: str) -> None:
     shutil.rmtree(pending.mint_root, ignore_errors=True)
 
 
+def republish_after_shape_warm(
+    pipe: Any,
+    cfg: Any,
+    family: str,
+    publisher: Optional["CellPublisher"],
+    live_root: Path,
+) -> bool:
+    """Replace the fleet cell with the live cache root's grown contents
+    (pgw#622): after a background novel-shape warm, the root holds the
+    adopted/minted graphs PLUS the fresh signature's, so republishing under
+    the same key means no other worker ever compiles that (shape, GPU,
+    lane) again. Synchronous (callers run it off the serving path); every
+    failure is non-fatal to serving."""
+    if publisher is None or not publisher.enabled():
+        logger.warning(
+            "hot-swap: SHAPE_WARM_WITHOUT_PUBLISH_SINK family=%s — the "
+            "novel-shape graphs stay local to this pod", family)
+        return False
+    if not (Path(live_root) / "inductor").is_dir():
+        logger.warning(
+            "hot-swap: live cache root %s has no inductor tree; nothing to "
+            "republish", live_root)
+        return False
+    from .models.loading import pipeline_weight_lane
+    from .models.memory import low_vram_mode
+
+    tmp_root = Path(tempfile.mkdtemp(prefix="cellrepub-"))
+    try:
+        graph_signature, weight_contract = cc.execution_contract(pipe, cfg)
+        meta = cc.artifact_metadata(
+            family=family,
+            source_ref="shape-warm",
+            shapes=cfg.shapes,
+            targets=cfg.targets,
+            guidance_scales=getattr(cfg, "guidance_scales", ()),
+            low_vram_mode=low_vram_mode(pipe),
+            compile_mode=(
+                "regional" if getattr(cfg, "regional", False) else "whole"),
+            weight_lane=pipeline_weight_lane(pipe),
+            lora_bucket=int(getattr(cfg, "lora_bucket", 0) or 0),
+            graph_signature=graph_signature,
+            weight_contract=weight_contract,
+        )
+        label = cc.flavor_label(
+            meta["sku"], meta["torch"], meta.get("weight_lane", ""))
+        artifact = cc.pack(Path(live_root), tmp_root / f"{label}.tar.gz", meta)
+        publisher.publish(family, artifact, meta)
+        return True
+    except CellPublishRefused as exc:
+        logger.warning("hot-swap: cell republish refused (hub decision): %s", exc)
+        return False
+    except Exception:
+        logger.warning(
+            "hot-swap: cell republish failed; the fleet keeps the previous "
+            "cell", exc_info=True)
+        return False
+    finally:
+        shutil.rmtree(tmp_root, ignore_errors=True)
+
+
 def abandon_self_mint(pending: "PendingSelfMint") -> None:
     """Discard a self-mint capture the proof did not certify (disproven or
     genuinely unexercised with no proven sibling). Never packed, never
@@ -636,5 +696,6 @@ def _cuda_ready() -> bool:
     "enable_compiled",
     "finalize_self_mint",
     "publish_self_mint",
+    "republish_after_shape_warm",
     "withhold_self_mint_publish",
 ]