pgw#626 / th#1059 twin: mandatory-lane admission follows the hub-resolved execution lane (0.50.1)

CHANGELOG.md

@@ -1,5 +1,19 @@
 # Changelog
 
+## 0.50.1 (2026-07-23)
+
+- **pgw#626 / th#1059 twin: mandatory-lane admission follows the hub-resolved
+  EXECUTION lane, not the flavor token.** The `#fp8-w8a8` flavor names the
+  STORAGE format; SDXL's mixed variant serves the w8a16 upcast lane (plain
+  graphs). `_validate_required_compile` (and every other mandatory-lane
+  derivation) refused hub dispatches for the mixed lane with
+  `required_compile_missing` — the worker half of the 2026-07-23 sdxl P0
+  (Paul's live jobs failed at 21:54Z). Mandatory-ness now derives from the
+  HelloAck resolution lane when known (w8a8/w4a4 activations stay
+  fail-closed; plain activations admit and JIT-warm like bf16); the flavor
+  token remains the fallback without lane evidence, and conflicting
+  evidence fails closed.
+
 ## 0.48.2 (2026-07-23)
 
 - **th#1043: the forced group-fit fp8 downgrade reports structurally.**

pyproject.toml

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

src/gen_worker/executor.py

@@ -2618,7 +2618,7 @@ def _install_compile_targets(
             else _CompileObjectCandidate(item, set(all_slots))
             for item in objects
         ]
-        requested_lane = _mandatory_lane_of(
+        requested_lane = self._mandatory_lane_of_bound(
             wire_ref(spec.models[slot]) for slot in self._setup_slots(spec)
         )
         active_artifacts = active_artifacts or {}
@@ -2723,7 +2723,7 @@ def _install_compile_targets(
             target_quant_lane = next(
                 (lane for lane in _MANDATORY_LANES
                  if target.pipeline_weight_lane.startswith(lane)), "")
-            candidate_requested_lane = _mandatory_lane_of(
+            candidate_requested_lane = self._mandatory_lane_of_bound(
                 ref for _slot, ref, _digest in bindings)
             mandatory_quant = bool(target_quant_lane)
             if (
@@ -2884,7 +2884,7 @@ def cell_lookups(self) -> List[pb.CellLookup]:
                 if cfg is None or not family:
                     continue
                 bucket = int(getattr(cfg, "lora_bucket", 0) or 0)
-                want_lane = _mandatory_lane_of(
+                want_lane = self._mandatory_lane_of_bound(
                     wire_ref(binding) for binding in spec.models.values()
                 )
                 lanes = (want_lane,) if want_lane else ("", "fp8-hooks")
@@ -2914,6 +2914,47 @@ def _compile_target(
                 return rec, target
         return None
 
+    def _resolved_mandatory_lane(self, ref: str) -> str:
+        """th#1059 twin (hub: ``mandatoryTracedLane``): the flavor token names
+        the STORAGE format, not the execution. Mandatory-ness follows the
+        hub-resolved EXECUTION lane whenever one is known for this ref —
+        SDXL's mixed variant is ``#fp8-w8a8`` storage serving the w8a16
+        upcast lane (plain graphs, never scaled_mm), while qwen's
+        ``#fp8-w8a8`` executes real w8a8. Without lane evidence the flavor
+        token remains the fallback; conflicting evidence fails closed to the
+        mandatory reading.
+        """
+        from .models import lanes as lanespec
+
+        ref = (ref or "").strip()
+        known = False
+        mandatory = ""
+        for declared, pick in (self._model_resolutions or {}).items():
+            resolved_ref = (pick[0] or declared).strip()
+            lane_str = (pick[2] or "").strip()
+            if not lane_str or ref not in (declared.strip(), resolved_ref):
+                continue
+            try:
+                lane = lanespec.parse_lane(lane_str)
+            except ValueError:
+                continue
+            known = True
+            if lane.activation == lanespec.ACT_W8A8:
+                mandatory = "w8a8"
+            elif lane.activation == lanespec.ACT_W4A4:
+                mandatory = "w4a4"
+        if known:
+            return mandatory
+        return _ref_mandatory_lane(ref)
+
+    def _mandatory_lane_of_bound(self, refs: typing.Iterable[str]) -> str:
+        """Resolution-aware :func:`_mandatory_lane_of` (th#1059)."""
+        for ref in refs:
+            lane = self._resolved_mandatory_lane(ref)
+            if lane:
+                return lane
+        return ""
+
     def _validate_required_compile(
         self, spec: EndpointSpec, run: pb.RunJob,
     ) -> None:
@@ -2925,7 +2966,7 @@ def _validate_required_compile(
         than execute on a merely same-family pipeline.
         """
         setup_slots = self._setup_slots(spec)
-        want_lane = _mandatory_lane_of(
+        want_lane = self._mandatory_lane_of_bound(
             wire_ref(spec.models[slot]) for slot in setup_slots
         )
         if not run.HasField("required_compile"):
@@ -3481,7 +3522,7 @@ async def ensure_setup(
                         rec.stale = True
                         break
             if rec.ready and not rec.stale and spec.compile is not None:
-                mandatory_lane = _mandatory_lane_of(
+                mandatory_lane = self._mandatory_lane_of_bound(
                     wire_ref(spec.models[slot])
                     for slot in self._setup_slots(spec)
                 )
@@ -5135,7 +5176,7 @@ async def _fetch_compile_snapshot(
         # checkpoints. Snapshot maps also contain attached cells and may carry
         # unrelated/prepositioned models, so they must not choose the lane.
         model_refs = [wire_ref(binding) for binding in spec.models.values()]
-        want_lane = _mandatory_lane_of(model_refs)
+        want_lane = self._mandatory_lane_of_bound(model_refs)
         want_bucket = int(getattr(spec.compile, "lora_bucket", 0) or 0)
         # th#883 pull-by-key: a key-flavored cell is selected only when its
         # key is one this runtime computed for itself (the same candidates

tests/test_mandatory_lane_th1059.py

@@ -0,0 +1,113 @@
+"""th#1059 twin (live master incident 2026-07-23): the `#fp8-w8a8` flavor
+token names the STORAGE format, not the execution. SDXL's mixed variant is
+`#fp8-w8a8` storage serving the w8a16 upcast lane (plain graphs), while
+qwen's `#fp8-w8a8` executes real scaled_mm w8a8. The worker's mandatory-lane
+admission (`_validate_required_compile`) refused every hub dispatch for the
+mixed lane with `required_compile_missing` — Paul's live jobs failed at
+21:54Z after the hub half (tensorhub th#1059) started dispatching them.
+
+Mandatory-ness must follow the hub-delivered resolution lane when known;
+the flavor token stays the fallback without lane evidence."""
+
+from __future__ import annotations
+
+import pytest
+
+from gen_worker import Compile, Resources
+from gen_worker.api.binding import Hub, wire_ref
+from gen_worker.api.errors import RetryableError
+from gen_worker.executor import Executor
+from gen_worker.pb import worker_scheduler_pb2 as pb
+from gen_worker.registry import EndpointSpec
+
+
+BARE = "acme/wai-illustrious:prod"
+MIXED = "acme/wai-illustrious:prod#fp8-w8a8"
+
+
+class _Resolutions:
+    """Just enough Executor surface for the resolution-aware lane methods."""
+
+    def __init__(self, resolutions):
+        self._model_resolutions = resolutions
+
+    _resolved_mandatory_lane = Executor._resolved_mandatory_lane
+    _mandatory_lane_of_bound = Executor._mandatory_lane_of_bound
+    _validate_required_compile = Executor._validate_required_compile
+    _setup_slots = staticmethod(Executor._setup_slots)
+
+
+class _Payload:
+    pass
+
+
+class _Endpoint:
+    def setup(self, checkpoint: str) -> None:  # pragma: no cover - shape only
+        pass
+
+    def run(self, ctx, payload):  # pragma: no cover - shape only
+        return None
+
+
+def _spec() -> EndpointSpec:
+    return EndpointSpec(
+        name="generate-turbo", method=_Endpoint.run, kind="inference",
+        payload_type=_Payload, output_mode="single", cls=_Endpoint,
+        attr_name="run",
+        models={"checkpoint": Hub("acme/wai-illustrious", tag="prod", flavor="fp8-w8a8")},
+        resources=Resources(vram_gb=1.0),
+        compile=Compile(family="sdxl", shapes=((1024, 1024),)),
+    )
+
+
+def test_w8a16_resolution_lane_is_not_mandatory() -> None:
+    ex = _Resolutions({BARE: (MIXED, "", "fp8-w8a16+compiled")})
+    assert ex._resolved_mandatory_lane(MIXED) == ""
+    assert ex._resolved_mandatory_lane(BARE) == ""
+    assert ex._mandatory_lane_of_bound([MIXED]) == ""
+
+
+def test_w8a8_resolution_lane_stays_mandatory() -> None:
+    ex = _Resolutions({BARE: (MIXED, "", "fp8-w8a8-dynamic+compiled")})
+    assert ex._resolved_mandatory_lane(MIXED) == "w8a8"
+    assert ex._mandatory_lane_of_bound([MIXED]) == "w8a8"
+
+
+def test_flavor_token_fallback_without_lane_evidence() -> None:
+    ex = _Resolutions({})
+    assert ex._resolved_mandatory_lane(MIXED) == "w8a8"
+    assert ex._resolved_mandatory_lane("acme/other:prod#nvfp4-w4a4") == "w4a4"
+    assert ex._resolved_mandatory_lane(BARE) == ""
+    empty_lane = _Resolutions({BARE: (MIXED, "", "")})
+    assert empty_lane._resolved_mandatory_lane(MIXED) == "w8a8"
+
+
+def test_conflicting_lane_evidence_fails_closed() -> None:
+    ex = _Resolutions({
+        BARE: (MIXED, "", "fp8-w8a16+compiled"),
+        "acme/alias:prod": (MIXED, "", "fp8-w8a8-dynamic+compiled"),
+    })
+    assert ex._resolved_mandatory_lane(MIXED) == "w8a8"
+
+
+def test_mixed_lane_dispatch_admits_without_required_compile() -> None:
+    """The live failure shape: RunJob without required_compile for the mixed
+    checkpoint must ADMIT (JIT setup), not raise required_compile_missing."""
+    ex = _Resolutions({BARE: (MIXED, "", "fp8-w8a16+compiled")})
+    spec = _spec()
+    run = pb.RunJob(
+        function_name=spec.name,
+        models=[pb.ModelBinding(slot="checkpoint", ref=wire_ref(spec.models["checkpoint"]))],
+    )
+    ex._validate_required_compile(spec, run)  # must not raise
+
+
+def test_w8a8_dispatch_without_required_compile_still_refuses() -> None:
+    ex = _Resolutions({BARE: (MIXED, "", "fp8-w8a8-dynamic+compiled")})
+    spec = _spec()
+    run = pb.RunJob(
+        function_name=spec.name,
+        models=[pb.ModelBinding(slot="checkpoint", ref=wire_ref(spec.models["checkpoint"]))],
+    )
+    with pytest.raises(RetryableError, match="required_compile_missing"):
+        ex._validate_required_compile(spec, run)