gw#603: custom-warmup proof attributes to contract-compatible siblings; warmup=None stays fail-closed

src/gen_worker/executor.py

@@ -2424,6 +2424,11 @@ def _install_compile_targets(
                 if id(pipeline) in function_proofs
                 else contract_names
             )
+            object_proven_by_custom_warmup = bool(
+                spec.cls is not None
+                and callable(getattr(spec.cls, "warmup", None))
+                and function_proofs.get(id(pipeline))
+            )
             incarnation_id = uuid.uuid4().hex
             target = _CompileTargetRecord(
                 incarnation_id=incarnation_id,
@@ -2469,6 +2474,28 @@ def _install_compile_targets(
                 if name:
                     compatible_names.add(name)
             expected_names = compatible_names & required_names
+            if object_proven_by_custom_warmup:
+                # gw#603 ruling (2026-07-20, supersedes the ac0bab9 single-
+                # name attribution): proof is a property of the WARMED
+                # OBJECT and the graph set actually exercised, not of the
+                # initiating handler's name — the same identity reasoning as
+                # gw#587 design pt 5. A custom object-level warmup (author
+                # surface 3: "wins outright", e.g. LTX's two-stage synthetic
+                # that warms EVERY declared graph) therefore attributes its
+                # proof to every CONTRACT-COMPATIBLE sibling alias of this
+                # exact object (same family, lora bucket, execution-contract
+                # digest, and bindings — the compatible_names gate above),
+                # EXCEPT aliases whose spec explicitly declares
+                # `warmup={...: None}`: those remain fail-closed (the
+                # SDXL-legacy-Turbo carve-out — an author's explicit "this
+                # handler must not be warmed/served on this lane" is a
+                # contract, never overridden by a sibling's proof). Runtime
+                # backstop stays: every advertised alias serves through the
+                # per-call guarded wrapper with hit/miss counters, so an
+                # attributed alias whose real requests miss is visible and
+                # degrades loudly, never silently.
+                permitted_names = (
+                    compatible_names - self._custom_warmup_optouts(spec))
             target.function_names = tuple(sorted(
                 compatible_names & permitted_names))
             target_quant_lane = next(
@@ -2479,8 +2506,11 @@ def _install_compile_targets(
             mandatory_quant = bool(target_quant_lane)
             if (
                 (mandatory_quant or candidate_requested_lane)
-                and set(target.function_names) != expected_names
+                and not expected_names <= set(target.function_names)
             ):
+                # Every REQUIRED alias must be proven; a proven superset
+                # (custom-warmup attribution covering a non-required
+                # sibling) is not a defect.
                 raise compile_cache.CompiledLaneUnavailableError(
                     "mandatory quantized-lane function proof incomplete "
                     f"(expected={sorted(expected_names)!r} "
@@ -3130,12 +3160,37 @@ def _warmup_plan(
             has_warmup_method=False,
         )
 
+    def _custom_warmup_optouts(self, spec: EndpointSpec) -> set[str]:
+        """Function names whose spec explicitly declares ``warmup={...: None}``.
+
+        gw#603: the explicit-None carve-out from custom-warmup proof
+        attribution — an author's declared skip is a per-handler contract
+        ("never warm/serve this alias on this lane"), so it stays fail-closed
+        even when the object-level proof would otherwise cover it."""
+        if spec.cls is None:
+            return set()
+        from .api.decorators import ATTR as _DECL_ATTR
+
+        decl = getattr(spec.cls, _DECL_ATTR, None)
+        declared = getattr(decl, "warmup", None)
+        if not isinstance(declared, typing.Mapping):
+            return set()
+        skipped_attrs = {a for a, v in declared.items() if v is None}
+        if not skipped_attrs:
+            return set()
+        return {
+            s.name for s in self.specs.values()
+            if s.cls is spec.cls and s.attr_name in skipped_attrs
+        }
+
     def _compile_contract_names(
         self, spec: EndpointSpec, rec: _ClassRecord,
     ) -> set[str]:
         """Handler aliases this setup can attribute its warmup proof to."""
         if spec.cls is not None and callable(getattr(spec.cls, "warmup", None)):
-            # A custom object-level warmup has no per-handler attribution.
+            # Absent a completed object proof, a custom object-level warmup
+            # has no per-handler attribution (the gw#603 attribution in
+            # _install_compile_targets applies only to a PROVEN object).
             return {spec.name}
         return self._required_compile_names(spec, rec)
 

tests/test_executor_adopt.py

@@ -2152,10 +2152,21 @@ async def _download(ref, **kwargs):
     assert ex.unavailable[spec.name][0] == "compile_cell_failed"
 
 
-def test_w8a8_partial_handler_proof_fails_loud_without_disabling_skipped_turbo(
+def test_w8a8_custom_warmup_proof_attributes_to_siblings_except_declared_none(
     tmp_path, monkeypatch,
 ):
-    """Every required alias must prove the W8A8 cell; explicit skips do not."""
+    """gw#603 ruling (2026-07-20, rewrites the ac0bab9 pin — see the tracker
+    note in gw#603 for the reversal rationale + gw#595's original reasoning):
+    proof is a property of the WARMED OBJECT and the graph set actually
+    exercised, not of the initiating handler's name. A custom object-level
+    warmup's proof attributes to every contract-compatible sibling alias of
+    the exact proven object — EXCEPT aliases explicitly declared
+    ``warmup={...: None}``, which stay fail-closed (the legacy-Turbo
+    carve-out; this exclusion is the revert-turns-red for the safeguard).
+    Live motivation: LTX serves generate+edit(+extend) from ONE class with
+    ONE custom warmup that warms every declared graph — under the ac0bab9
+    single-name attribution no >=0.38.8 worker could EVER boot it compiled,
+    delivered cells included."""
     import gen_worker.executor as executor_mod
 
     family = "sdxl"
@@ -2214,20 +2225,93 @@ async def _download(ref, **kwargs):
     )
     monkeypatch.setattr(ex, "_enable_compiled", _guarded_enable)
 
-    with pytest.raises(
-        cc.CompiledLaneUnavailableError,
-        match="mandatory quantized-lane function proof incomplete",
-    ):
-        asyncio.run(ex.ensure_setup(generate, {
-            model_ref: pb.Snapshot(digest=MODEL_DIGEST),
-            cell_ref: pb.Snapshot(digest=DIGEST_A),
-        }))
+    instance = asyncio.run(ex.ensure_setup(generate, {
+        model_ref: pb.Snapshot(digest=MODEL_DIGEST),
+        cell_ref: pb.Snapshot(digest=DIGEST_A),
+    }))
+    assert isinstance(instance, _SdxlEndpoint)
 
-    required = {by_attr["generate"].name, by_attr["edit"].name}
-    assert set(ex.unavailable) == required
-    assert all(ex.unavailable[name][0] == "compile_cell_failed" for name in required)
-    assert by_attr["generate_turbo"].name not in ex.unavailable
-    assert ex.compile_targets() == []
+    # The object proof covers both compatible siblings...
+    (target,) = ex.compile_targets()
+    proven = set(target.function_names)
+    assert by_attr["generate"].name in proven
+    assert by_attr["edit"].name in proven
+    # ...but NEVER the explicitly opted-out alias (revert-turns-red for the
+    # warmup={...: None} carve-out): Turbo rejects W8A8 by author contract
+    # and must not be advertised as servable on this lane.
+    assert by_attr["generate_turbo"].name not in proven
+    assert not ex.unavailable
+
+
+def test_w8a8_custom_warmup_multi_alias_boot_serves_all_siblings(
+    tmp_path, monkeypatch,
+):
+    """The exact live gw#603 shape (LTX): one class, generate+edit aliases,
+    ONE custom warmup() covering every declared graph, NO decorator warmup
+    rows. Under single-name attribution this boot failed closed forever
+    ("expected=['edit','generate'] proven=['edit']") on delivered AND
+    self-mint cells alike; under the gw#603 ruling the proven object
+    certifies both siblings and the boot serves."""
+    import gen_worker.executor as executor_mod
+
+    family = "ltx-shaped"
+    cell_ref = f"_system/family-{family}#inductor-rtx-4090-torch2.9-w8a8"
+    artifact = _artifact(tmp_path, family=family)
+    model_dir = tmp_path / "ltx-shaped-model"
+    model_dir.mkdir()
+
+    @endpoint(
+        models={"pipeline": Hub("acme/ltx-shaped", flavor="fp8-w8a8")},
+        resources=Resources(vram_gb=24),
+        compile=Compile(shapes=((1024, 1024),), family=family),
+    )
+    class _LtxShapedEndpoint:
+        def setup(self, pipeline: _LoadablePipe) -> None:
+            self.pipeline = pipeline
+
+        def warmup(self) -> None:
+            # The instance-level synthetic warms EVERY declared graph.
+            _record_fake_warm(self.pipeline)
+
+        def generate(self, ctx, payload: _In) -> _Out:
+            return _Out(y="ok")
+
+        def edit(self, ctx, payload: _In) -> _Out:
+            return _Out(y="ok")
+
+    specs = extract_specs(_LtxShapedEndpoint)
+    by_attr = {spec.attr_name: spec for spec in specs}
+    generate = by_attr["generate"]
+    model_ref = wire_ref(generate.models["pipeline"])
+    pipe = _LoadablePipe()
+    setattr(pipe, "_cozy_weight_lane", "w8a8")
+
+    async def _send(_msg):
+        return None
+
+    ex = Executor(specs, _send)
+    ex.store._cache_dir = tmp_path / "cas"
+
+    async def _download(ref, **kwargs):
+        return artifact.parent if ref == cell_ref else model_dir
+
+    monkeypatch.setattr(executor_mod, "ensure_local", _download)
+    monkeypatch.setattr(
+        provision,
+        "load_slot",
+        lambda *args, **kwargs: provision.SlotLoad(obj=pipe, is_pipeline=True),
+    )
+    monkeypatch.setattr(ex, "_enable_compiled", _guarded_enable)
+
+    instance = asyncio.run(ex.ensure_setup(generate, {
+        model_ref: pb.Snapshot(digest=MODEL_DIGEST),
+        cell_ref: pb.Snapshot(digest=DIGEST_A),
+    }))
+    assert isinstance(instance, _LtxShapedEndpoint)
+    (target,) = ex.compile_targets()
+    assert set(target.function_names) == {
+        by_attr["generate"].name, by_attr["edit"].name}
+    assert not ex.unavailable
 
 
 def _merged_lane_endpoint(record_warm):