From edefbd437332c52304dc24c9c08093793cfe24a6 Mon Sep 17 00:00:00 2001 From: Paul Fidika Date: Tue, 11 Aug 2026 12:48:45 -0600 Subject: [PATCH] =?UTF-8?q?pgw#1139=20(ie#657):=20the=20flow=20gate=20read?= =?UTF-8?q?s=20scheduler=20CAPABILITY,=20not=20the=20class=20name=20?= =?UTF-8?q?=E2=80=94=200.109.0=20->=200.110.0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `clone_scheduler` admitted an objective="flow" checkpoint only when `cls.__name__.startswith("FlowMatch")`. The official Wan solver is `UniPCMultistepScheduler(prediction_type="flow_prediction", use_flow_sigmas=True, flow_shift=5.0)` — upstream's own scheduler_config.json, byte-identical in our mirrors — so every request against it raised before the first denoise step, and the fatal requeued onto a fresh H100 per attempt. Measured blast radius on the master stack: wan-2.2's `text-image-to-video` has ONE request row ever (cancelled), `video-to-video` and `image-to-video` have none. A name test refuses a config-declared flow sampler for EVERY consumer, so the fix is the gate, not a third endpoint-local `_serving_view` bypass. `flow_capable(cls, config)` reads the facts instead: flow by construction (the `FlowMatch*` family) or flow by declaration — `use_flow_sigmas` / `prediction_type="flow_prediction"` in the RESOLVED config AND a class that takes `use_flow_sigmas`. The capability half is load-bearing: every diffusers scheduler accepts `prediction_type`, so a declaration alone would let a flow config carried onto EulerAncestral read as flow while from_config dropped it. What the gate still refuses is unchanged, and now tested both ways: a diffusion sampler over a flow checkpoint, a flow-CAPABLE class its config never declares, a non-diffusers scheduler declaring nothing. Three new tests are red on the pre-fix tree — the acceptance one with the production error verbatim — and the three refusal tests pass on both sides. ie#535's second half rides along: `shift` and `flow_shift` are the same knob under two upstream spellings and from_config drops the one the class does not take, so a published shift was ignored rather than applied. A declared shift is renamed to the spelling the class honours; nothing is invented. The cut carries pgw#1132 as well (merged after 0.109.0 was already published). --- CHANGELOG.md | 12 ++++ changelog.d/pgw1132.md | 1 - pyproject.toml | 2 +- src/gen_worker/view.py | 99 ++++++++++++++++++++++++++---- tests/test_objectives_pgw654.py | 103 ++++++++++++++++++++++++++++++++ uv.lock | 2 +- 6 files changed, 203 insertions(+), 16 deletions(-) delete mode 100644 changelog.d/pgw1132.md diff --git a/CHANGELOG.md b/CHANGELOG.md index da3540b2..0bd068b2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,18 @@ Unreleased entries live in [`changelog.d/`](changelog.d/), one file per issue — lanes do not edit this file. A cut assembles them (`scripts/assemble_changelog.py`). +## 0.110.0 (2026-08-11) — **the flow-objective gate reads scheduler CAPABILITY, not the class name** — three wan-2.2 functions could not execute a single request + +- **pgw#1132: the boot-key derivation arms the LIFTED forward it is about to export — without it no `lora_bucket` family could ever ask for its cell.** pgw#822 at the mint was the child arming the branch CONTAINERS and handing `torch.export` the bare denoiser; the fix was `mint_targets` owning its arm (`_arm_branches`, before the first export). §4.27 step 1's loop, `aot_mint.trace_for_key`, was left with that call **only in its `finally`** — the re-arm after the branchless group, never before the first row — while its driver `boot_trace_child.run` arms the container half alone (`cc.apply_lora_execution_lane`, exactly as `mint_child` does). So the FIRST adapter-bearing row of every bucket-bearing family met `_export_entry`'s pgw#822 gate on a container-only pipeline and refused verbatim (*"this class declares lifted adapter input(s) ['lora_a', 'lora_b'] but ZImageTransformer2DModel carries no lifted forward"*), `boot_trace_child` reported `trace_refused`, and the whole derivation died **before a resolve was possible** — AOT adoption was 100 % unreachable for `qwen-image` t2i and `qwen-image-edit` (128), `z-image` (128), `wan-2.2` (128/64), `sdxl` (64), `sd15` (64) and `anima` (32), including one of the four workhorses. One call, mirroring `mint_targets`: the loop arms itself before the row loop, per share, and the caller-side container arm stays the caller's. `tests/test_boot_key_lora_arm_pgw1132.py` derives both fork arms of a bucket-bearing declaration through the REAL `trace_for_key` over the REAL `_export_entry` gate — nothing stubbed, five rows red before the arm — and holds the fork apart (`lifted_inputs` present on `adapter=true`, empty on `adapter=false`), the per-share arm, and the loop's post-condition that the pipeline is left on the lifted family. + +- **pgw#1139 (ie#657): the flow-objective gate tested a class NAME and darkened three serving functions.** `view.clone_scheduler` admitted an `objective="flow"` checkpoint only when `cls.__name__.startswith("FlowMatch")`. The official Wan solver is `UniPCMultistepScheduler` with `prediction_type="flow_prediction"`, `use_flow_sigmas=true`, `flow_shift=5.0` — upstream's own `scheduler/scheduler_config.json`, byte-identical in our mirrors — so every request against it raised verbatim *"objective='flow' checkpoint cannot run under sampler 'UniPCMultistepScheduler' (UniPCMultistepScheduler is not a flow-match scheduler)"* **before the first denoise step**, and a fatal that requeues bought a fresh H100 per attempt. Realized blast radius, measured off the master stack's whole request table: wan-2.2's `text-image-to-video` had ONE row ever (cancelled), `video-to-video` and `image-to-video` had NONE — no TI2V-5B frame has ever been rendered. Only the `-turbo` siblings escaped, through an endpoint-local `_serving_view` that bypasses the SDK entirely; the honest fix is one gate, not a third bypass, because a name test refuses a config-declared flow sampler for **every** consumer. + + New `view.flow_capable(cls, config)` reads the two facts that are actually true or false: flow BY CONSTRUCTION (the `FlowMatch*` family has no other mode) or flow BY DECLARATION — the resolved config says `use_flow_sigmas` / `prediction_type="flow_prediction"` **and** the class takes `use_flow_sigmas` at all. The second half is the whole check: every diffusers scheduler accepts `prediction_type`, so a declaration alone would let a flow config carried onto `EulerAncestralDiscreteScheduler` read as flow while `from_config` silently dropped it. **What the gate still refuses is unchanged and now tested in both directions**: a diffusion sampler selected over a flow checkpoint (`sampler="euler_a"`), a flow-capable class left UNDECLARED by its config (plain UniPC is a diffusion sampler), and a non-diffusers scheduler that declares nothing. + + **ie#535's second half, closed with it.** `shift` and `flow_shift` are the same flow-sigma knob under two upstream spellings (`FlowMatch*` vs the multistep solvers), and `from_config` DROPS the one the target class does not take — so a repo publishing a flow shift onto a UniPC mirror had it ignored rather than refused, silently. A declared shift is now renamed to whichever spelling the class honours; nothing is invented, and a class taking neither is left alone. + + Revert-turns-red: `tests/test_objectives_pgw654.py` builds the real `UniPCMultistepScheduler` from the **verbatim** upstream Wan config and drives the real `clone_scheduler`. Three tests are RED on the pre-fix tree — the acceptance test failing with the production error string character-for-character — while the three refusal tests pass on both sides, which is what proves the fix did not simply open the gate. + ## 0.109.0 (2026-08-11) — **a dtype cast respects per-component precision pins** — the number 0.108.0 was already spent on a tree that does not carry this - **pgw#1133: a dtype cast is tree-wide, a precision pin is per-component — and only the LOAD half knew it.** `families.facts` states `AutoencoderKLWan -> fp32` (*"the Wan VAE is numerically fragile: bf16 latents degrade decoded frames visibly"*) and `models/loading.py` honours it on every materialize. `convert/clone.py`'s cast did not consult it at all — `quantize_components` is reached only when `is_quant`, so a plain `dtype: "bf16"` clone applied to **every** weight group in the tree. Live, on the master stack: ie#649 leg A published `vae/diffusion_pytorch_model.safetensors` at **253,806,966 B** against the **507,591,892 B** fp32 upstream, valid and classified and complete, and a serving pod would have upcast the truncation back into fp32 with nothing anywhere saying so. New `convert/dtype_pins.py` is the producer half of that one fact: it reads the TREE'S OWN `model_index.json` classes and asks `families.facts` — never a second table. A tree-wide NARROWING cast **skips** the pinned component (passthrough at source precision) and reports it in `attr_dtype_pinned_components`, because "publish a bf16 flavor of this model" has no per-component spelling that could satisfy both the request and the pin, and refusing would leave the operator with ie#649's third pod leg as the only path to the same tree; a cast that WIDENS is a no-op and is not touched. An **explicit** instruction naming a pinned component as a quant target is refused instead, typed and by name (`ComponentDtypePinError`), because that one has a repair. Two publish gates close it behind the cast: `run_clone` and `publish_flavors` both refuse a tree carrying a pinned component narrower than its pin (`ComponentDtypePinViolation`) — the clone arm compares against the SOURCE, so an upstream that ships the component narrow is still mirrorable and only OUR truncation fails. Both publish `metadata.component_dtypes`, so per-component precision is a queryable checkpoint fact instead of a byte count somebody has to notice. **The catalog audit that followed found the defect had already shipped**: `tensorhub/wan22-ti2v-5b`'s `prod`+`latest` head carries a **1,409,400,600 B bf16** `AutoencoderKLWan` against upstream's **2,818,777,808 B** fp32 (196 tensors, all BF16 by header) — a whole-tree fp32→bf16 cast whose transformer is likewise exactly half of upstream. Reported for its owner to re-clone rather than silently re-cast; the two ie#649 intermediates carrying the same downcast are untagged and superseded. diff --git a/changelog.d/pgw1132.md b/changelog.d/pgw1132.md deleted file mode 100644 index 3bda8bb9..00000000 --- a/changelog.d/pgw1132.md +++ /dev/null @@ -1 +0,0 @@ -- **pgw#1132: the boot-key derivation arms the LIFTED forward it is about to export — without it no `lora_bucket` family could ever ask for its cell.** pgw#822 at the mint was the child arming the branch CONTAINERS and handing `torch.export` the bare denoiser; the fix was `mint_targets` owning its arm (`_arm_branches`, before the first export). §4.27 step 1's loop, `aot_mint.trace_for_key`, was left with that call **only in its `finally`** — the re-arm after the branchless group, never before the first row — while its driver `boot_trace_child.run` arms the container half alone (`cc.apply_lora_execution_lane`, exactly as `mint_child` does). So the FIRST adapter-bearing row of every bucket-bearing family met `_export_entry`'s pgw#822 gate on a container-only pipeline and refused verbatim (*"this class declares lifted adapter input(s) ['lora_a', 'lora_b'] but ZImageTransformer2DModel carries no lifted forward"*), `boot_trace_child` reported `trace_refused`, and the whole derivation died **before a resolve was possible** — AOT adoption was 100 % unreachable for `qwen-image` t2i and `qwen-image-edit` (128), `z-image` (128), `wan-2.2` (128/64), `sdxl` (64), `sd15` (64) and `anima` (32), including one of the four workhorses. One call, mirroring `mint_targets`: the loop arms itself before the row loop, per share, and the caller-side container arm stays the caller's. `tests/test_boot_key_lora_arm_pgw1132.py` derives both fork arms of a bucket-bearing declaration through the REAL `trace_for_key` over the REAL `_export_entry` gate — nothing stubbed, five rows red before the arm — and holds the fork apart (`lifted_inputs` present on `adapter=true`, empty on `adapter=false`), the per-share arm, and the loop's post-condition that the pipeline is left on the lifted family. diff --git a/pyproject.toml b/pyproject.toml index 679b97f5..83caae86 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "gen-worker" -version = "0.109.0" +version = "0.110.0" description = "A library used to build custom functions in Cozy Creator's serverless function platform." readme = "README.md" license = "MIT" diff --git a/src/gen_worker/view.py b/src/gen_worker/view.py index 02ce31f2..21419c9b 100644 --- a/src/gen_worker/view.py +++ b/src/gen_worker/view.py @@ -47,6 +47,7 @@ from __future__ import annotations import copy +import inspect from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple # Friendly sampler name -> (diffusers scheduler class name, extra config). @@ -96,16 +97,78 @@ "unipc": ("UniPCMultistepScheduler", {}), } -# Scheduler classes that integrate a FLOW-MATCHING objective. A checkpoint -# stamped objective="flow" must never be driven by a diffusion (eps/v-pred) -# scheduler — the sigma schedules are different math, not a preference. +# A checkpoint stamped objective="flow" must never be driven by a diffusion +# (eps/v-pred) sampler — the sigma schedules are different math, not a +# preference. What makes a scheduler flow-match is CAPABILITY, not spelling: +# the `FlowMatch*` family is flow by construction, and the multistep solvers +# (UniPC, DPMSolver, DEIS) are flow when their config says so. UniPC with +# `use_flow_sigmas` IS the official Wan solver (pgw#1139/ie#657). _FLOW_CLASS_PREFIX = "FlowMatch" +# The __init__ field that only a flow-capable non-FlowMatch class has. +_FLOW_SIGMAS_FIELD = "use_flow_sigmas" +_FLOW_PREDICTION_TYPE = "flow_prediction" +# The same flow-sigma knob under two upstream spellings: `FlowMatch*` calls it +# `shift`, the multistep solvers call it `flow_shift`. +_FLOW_SHIFT_ALIASES = ("shift", "flow_shift") class UnknownSamplerError(ValueError): """The requested sampler name is not in the SDK sampler table.""" +def _init_fields(cls: Any) -> frozenset: + """The config fields ``cls`` actually accepts. ``from_config`` silently + DROPS everything else, so this is the only honest "does it honour it".""" + if not isinstance(cls, type): + return frozenset() + try: + return frozenset(inspect.signature(cls).parameters) + except (TypeError, ValueError): # pragma: no cover - exotic callable + return frozenset() + + +def _as_dict(config: Any) -> Dict[str, Any]: + """A diffusers ``FrozenDict`` as a plain dict; anything unreadable as {}.""" + try: + return dict(config or {}) + except (TypeError, ValueError): + return {} + + +def flow_capable(cls: Any, config: Optional[Dict[str, Any]] = None) -> bool: + """Does the scheduler this class+config BUILDS integrate flow-match sigmas? + + Two honest signals, and a class name is only one of them: + + * flow BY CONSTRUCTION — the ``FlowMatch*`` family has no other mode; + * flow BY DECLARATION — the resolved config says + ``use_flow_sigmas`` / ``prediction_type="flow_prediction"``, AND the + class takes ``use_flow_sigmas`` at all. The second half is the whole + check: every diffusers scheduler accepts ``prediction_type``, so a flow + config carried onto a diffusion class would otherwise read as flow while + producing a scheduler that merely looked flow-shaped in a dict. + """ + name = cls.__name__ if isinstance(cls, type) else str(cls) + if name.startswith(_FLOW_CLASS_PREFIX): + return True + if _FLOW_SIGMAS_FIELD not in _init_fields(cls): + return False + cfg = config or {} + return bool(cfg.get(_FLOW_SIGMAS_FIELD)) or \ + cfg.get("prediction_type") == _FLOW_PREDICTION_TYPE + + +def _alias_flow_shift(cls: Any, overrides: Dict[str, Any]) -> None: + """Rename a declared flow shift to the field ``cls`` honours (ie#535's + second half): a published ``{"shift": …}`` on a UniPC mirror is dropped by + ``from_config`` — ignored rather than refused. Only renames when the class + takes exactly one of the two spellings; never invents a value.""" + fields = _init_fields(cls) + for key, other in (_FLOW_SHIFT_ALIASES, _FLOW_SHIFT_ALIASES[::-1]): + if key in overrides and key not in fields and other in fields: + overrides[other] = overrides.pop(key) + + def _scheduler_class(sampler: str) -> Tuple[Any, Dict[str, Any]]: key = str(sampler or "").strip().lower() if key not in SAMPLERS: @@ -211,11 +274,18 @@ def clone_scheduler( v-prediction ALSO sets ``rescale_betas_zero_snr=True`` (th#1017's zero-terminal-SNR contract — folded in here so no endpoint can forget it and wash out). - - ``"flow"``: requires a flow-match scheduler class — a diffusion + - ``"flow"``: requires a scheduler that INTEGRATES flow-match sigmas + (:func:`flow_capable` — flow by construction, or a class that takes + ``use_flow_sigmas`` under a config that declares it); a diffusion sampler selection raises instead of silently integrating the wrong - math; the sigma schedule rides the instance scheduler's config. + math, and the sigma schedule rides the instance scheduler's config. - ``""`` (unstamped): applies nothing. + A declared flow ``shift`` is renamed to whichever of + ``shift``/``flow_shift`` the target class honours — ``from_config`` drops + the other spelling, so a published shift would be ignored rather than + applied (ie#535). + ``attr`` names which scheduler attribute to clone (default the primary ``scheduler``); :func:`for_request` uses it for secondary samplers. """ @@ -236,15 +306,18 @@ def clone_scheduler( overrides = {**extra, **overrides} else: cls = type(base) - if obj == "flow": + _alias_flow_shift(cls, overrides) + resolved = {**_as_dict(base_config), **overrides} + if obj == "flow" and not flow_capable(cls, resolved): cls_name = cls.__name__ if isinstance(cls, type) else str(cls) - if not cls_name.startswith(_FLOW_CLASS_PREFIX): - raise ValueError( - f"objective='flow' checkpoint cannot run under sampler " - f"{sampler or cls_name!r} ({cls_name} is not a flow-match " - "scheduler); flow sigma schedules are different math, not a " - "preference" - ) + raise ValueError( + f"objective='flow' checkpoint cannot run under sampler " + f"{sampler or cls_name!r} ({cls_name} integrates no flow-match " + f"sigmas: not a {_FLOW_CLASS_PREFIX}* class, and the resolved " + f"scheduler config declares no {_FLOW_SIGMAS_FIELD} / " + f"prediction_type={_FLOW_PREDICTION_TYPE!r} that it accepts); " + "flow sigma schedules are different math, not a preference" + ) if base_config is None or not hasattr(cls, "from_config"): # Non-diffusers scheduler shape: fall back to a deepcopy (still a # private per-request object; state never shared). diff --git a/tests/test_objectives_pgw654.py b/tests/test_objectives_pgw654.py index a32a0202..ffe35901 100644 --- a/tests/test_objectives_pgw654.py +++ b/tests/test_objectives_pgw654.py @@ -422,6 +422,109 @@ def test_flow_objective_refuses_a_diffusion_sampler() -> None: clone_scheduler(_StubPipe(_StubScheduler()), sampler="euler_a", objective="flow") +# --------------------------------------------------------------------------- +# pgw#1139 / ie#657: the flow gate reads CAPABILITY, not the class NAME. +# +# `WAN_SCHEDULER_CONFIG` is verbatim +# `Wan-AI/Wan2.2-TI2V-5B-Diffusers@b8fff731:scheduler/scheduler_config.json` — +# the official Wan solver, which our mirrors ship byte-identically. It is a +# UniPC, so a `cls.__name__.startswith("FlowMatch")` test refused every +# `objective="flow"` request against it (three wan-2.2 functions, zero renders +# ever). +# --------------------------------------------------------------------------- + +WAN_SCHEDULER_CONFIG = { + "_class_name": "UniPCMultistepScheduler", + "beta_end": 0.02, "beta_schedule": "linear", "beta_start": 0.0001, + "disable_corrector": [], "dynamic_thresholding_ratio": 0.995, + "final_sigmas_type": "zero", "flow_shift": 5.0, "lower_order_final": True, + "num_train_timesteps": 1000, "predict_x0": True, + "prediction_type": "flow_prediction", "rescale_betas_zero_snr": False, + "sample_max_value": 1.0, "solver_order": 2, "solver_p": None, + "solver_type": "bh2", "steps_offset": 0, "thresholding": False, + "time_shift_type": "exponential", "timestep_spacing": "linspace", + "trained_betas": None, "use_beta_sigmas": False, + "use_dynamic_shifting": False, "use_exponential_sigmas": False, + "use_flow_sigmas": True, "use_karras_sigmas": False, +} + + +def _wan_pipe(): + from diffusers import UniPCMultistepScheduler + + return _StubPipe(UniPCMultistepScheduler.from_config(WAN_SCHEDULER_CONFIG)) + + +def test_flow_objective_accepts_a_config_declared_unipc() -> None: + """The official Wan solver SERVES: flow by config, not by class name.""" + pytest.importorskip("diffusers") + from diffusers import UniPCMultistepScheduler + from gen_worker.view import clone_scheduler + + fresh = clone_scheduler(_wan_pipe(), objective="flow") + assert isinstance(fresh, UniPCMultistepScheduler) + assert fresh.config.use_flow_sigmas is True + assert fresh.config.prediction_type == "flow_prediction" + + +def test_flow_objective_still_refuses_a_diffusion_sampler_on_that_tree() -> None: + """The other direction, on the SAME flow checkpoint: euler_a integrates no + flow sigmas, so the gate must still refuse it.""" + pytest.importorskip("diffusers") + from gen_worker.view import clone_scheduler + + with pytest.raises(ValueError, match="flow"): + clone_scheduler(_wan_pipe(), sampler="euler_a", objective="flow") + + +def test_flow_objective_refuses_a_flow_capable_class_left_undeclared() -> None: + """UniPC is flow-CAPABLE but not flow by construction: on a checkpoint whose + config never declares it, plain UniPC is a diffusion sampler and is refused + — the capability test is not a blanket pass for the class.""" + pytest.importorskip("diffusers") + from diffusers import UniPCMultistepScheduler + from gen_worker.view import clone_scheduler + + diffusion = dict(WAN_SCHEDULER_CONFIG, + prediction_type="epsilon", use_flow_sigmas=False) + pipe = _StubPipe(UniPCMultistepScheduler.from_config(diffusion)) + with pytest.raises(ValueError, match="flow"): + clone_scheduler(pipe, objective="flow") + + +def test_flow_capable_is_not_satisfied_by_prediction_type_alone() -> None: + """Every diffusers scheduler ACCEPTS `prediction_type`, so a declaration + carried onto a diffusion class must not read as flow.""" + pytest.importorskip("diffusers") + from diffusers import (EulerAncestralDiscreteScheduler, + FlowMatchEulerDiscreteScheduler, + UniPCMultistepScheduler) + from gen_worker.view import flow_capable + + declared = {"prediction_type": "flow_prediction", "use_flow_sigmas": True} + assert flow_capable(EulerAncestralDiscreteScheduler, declared) is False + assert flow_capable(UniPCMultistepScheduler, declared) is True + assert flow_capable(FlowMatchEulerDiscreteScheduler, {}) is True + + +def test_declared_flow_shift_reaches_the_field_the_class_honours() -> None: + """ie#535's second half: `{"shift": …}` on a UniPC mirror was DROPPED by + from_config — a published shift ignored rather than applied.""" + pytest.importorskip("diffusers") + from diffusers import FlowMatchEulerDiscreteScheduler + from gen_worker.view import clone_scheduler + + fresh = clone_scheduler(_wan_pipe(), objective="flow", + config_overrides={"shift": 3.0}) + assert fresh.config.flow_shift == 3.0 + + # and the reverse spelling, on a class whose field is `shift` + fm = _StubPipe(FlowMatchEulerDiscreteScheduler()) + back = clone_scheduler(fm, objective="flow", + config_overrides={"flow_shift": 7.0}) + assert back.config.shift == 7.0 + + def test_sampler_table_defines_euler_trailing_and_dpmpp_completely() -> None: # pgw#647 gap #2 absorbed: the SDK table is the ONE definition of each # named sampler; endpoints delete their private maps. diff --git a/uv.lock b/uv.lock index afceeebc..3277aea1 100644 --- a/uv.lock +++ b/uv.lock @@ -592,7 +592,7 @@ wheels = [ [[package]] name = "gen-worker" -version = "0.109.0" +version = "0.110.0" source = { editable = "." } dependencies = [ { name = "blake3" },