[bugfix] Fix extra work in VSA64 layout benchmark (#2027)

BaseLayerWithLoRA.forward was decorated with torch.compile, so wrapping a linear also compiled its base layer.
Compiling a packed NVFP4/FP8 linear on its own changes its numerics, so a wrapped layer at strength 0 could differ
from the unwrapped layer. forward now adds no compiled region: the base layer runs exactly as it would unwrapped
(eagerly, or traced with a caller's compiled block), and only the adapter delta is compiled.

docs/design/server_contracts/openai.md

@@ -88,27 +88,12 @@ block. If `num_frames` is absent, `seconds * fps` is used. FastVideo also keeps
 the legacy `input_reference`, `reference_url`, `video_path`, and `video_url`
 spellings.
 
-Reference objects carry one string each through `image_url`, `video_url`, and
-`audio_url`. Each field accepts these forms:
-
-| Field | `http(s)` URL | `data:` URL | Local path |
-| --- | --- | --- | --- |
-| `image_reference[].image_url`, `input_reference`, `reference_url` | yes | `data:image/...;base64,` | under `server.media_roots` |
-| `video_reference[].video_url`, `video_path`, `video_url` | yes | `data:video/...;base64,` | under `server.media_roots` |
-| `audio_reference[].audio_url` | yes | `data:audio/...;base64,` | under `server.media_roots` |
-
-`input_reference` also accepts a multipart upload. `file_id` references are
-schema-compatible but return HTTP 400 because FastVideo does not provide an
-OpenAI Files store.
-
-URLs must resolve to public addresses unless
-`server.allow_private_media_urls` is set. A `data:` URL must be base64-encoded,
-and its media type must match its field: a `data:audio/...` URL in
-`video_reference` is refused. Each downloaded file, and each decoded video or
-audio file, is limited to `server.media_reference_max_bytes`. URLs and `data:` URLs are
-written under `server.output_dir/uploads` during admission, keep their order
-within each field, and are deleted when the job ends. Invalid, oversized, or
-mismatched media returns HTTP 400 before a job is created.
+Reference objects support URL or local-path strings through `image_url`,
+`video_url`, and `audio_url`. `file_id` references are schema-compatible but
+return HTTP 400 because FastVideo does not provide an OpenAI Files store.
+Image URLs, data URLs, local paths, and multipart `input_reference` uploads are
+materialized and decoded under the configured output directory during
+admission. Invalid media returns HTTP 400 before a job is created.
 
 ## Jobs and synchronous responses
 

fastvideo-kernel/benchmarks/bench_vsa_tile_layout.py

@@ -5,7 +5,7 @@ Benchmark the SM100 fused VSA64 tile layout against the legacy scatter/transpose
 Times Wan VSA inference preprocessing through the backend
 (``VideoSparseAttentionImpl.preprocess_qkv``), toggling
 ``FASTVIDEO_DISABLE_VSA64_FUSED_LAYOUT``:
-  - legacy: tile scatter into BSHD + four BSHD->BHSD transposes
+  - legacy: tile scatter into BSHD + three contiguous Q/K/V transposes + a gate view
   - fused:  one Triton kernel writing tiled BHSD directly
 
 With ``--include_attn`` the timed region also covers ``forward`` (the
@@ -59,20 +59,20 @@ def parse_arguments() -> argparse.Namespace:
 
 
 def make_step(impl: VideoSparseAttentionImpl, qkvg: torch.Tensor, builder: VideoSparseAttentionMetadataBuilder,
-              build_kwargs: dict, fused: bool, include_attn: bool) -> Callable[[], torch.Tensor]:
+              build_kwargs: dict, fused: bool, include_attn: bool) -> Callable[[], tuple[torch.Tensor, ...]]:
     metadata = builder.build(**build_kwargs)
 
-    def step() -> torch.Tensor:
+    def step() -> tuple[torch.Tensor, ...]:
         # The routing gate reads the switch on every call.
         with envs.FASTVIDEO_DISABLE_VSA64_FUSED_LAYOUT.override(not fused):
             tiled = impl.preprocess_qkv(qkvg, metadata)
         q, k, v, gate = tiled.chunk(4)
         if include_attn:
-            return impl.forward(q, k, v, gate, metadata)
+            return (impl.forward(q, k, v, gate, metadata), )
         if metadata.fused_layout_active:
-            return tiled
-        # Legacy route: the transposes forward() applies before video_sparse_attn.
-        return torch.cat([t.transpose(1, 2).contiguous() for t in (q, k, v, gate)])
+            return q, k, v, gate
+        return (q.transpose(1, 2).contiguous(), k.transpose(1, 2).contiguous(), v.transpose(1, 2).contiguous(),
+                gate.transpose(1, 2))
 
     with torch.no_grad():
         step()
@@ -100,7 +100,7 @@ def main() -> None:
     legacy = make_step(impl, qkvg, builder, build_kwargs, fused=False, include_attn=args.include_attn)
     fused = make_step(impl, qkvg, builder, build_kwargs, fused=True, include_attn=args.include_attn)
     with torch.no_grad():
-        if not torch.equal(legacy(), fused()):
+        if not all(torch.equal(old, new) for old, new in zip(legacy(), fused(), strict=True)):
             raise RuntimeError("Fused and legacy outputs differ.")
         legacy_ms = do_bench(legacy, warmup=args.warmup, rep=args.rep, return_mode="median")
         fused_ms = do_bench(fused, warmup=args.warmup, rep=args.rep, return_mode="median")

fastvideo-kernel/tests/test_vsa_tile_layout.py

@@ -5,6 +5,9 @@ Production only routes SM100 here, but the kernel is plain Triton, so parity is
 checked on any CUDA device.
 """
 
+import importlib.util
+from pathlib import Path
+
 import pytest
 import torch
 
@@ -40,6 +43,63 @@ def test_tile_to_bhsd_matches_scatter_and_four_transposes(dit_shape):
     source_index[metadata.non_pad_index] = metadata.tile_partition_indices.to(torch.int32)
     out = torch.empty((4, 2, padded_sequence, 128), device="cuda", dtype=torch.bfloat16)
     actual = tile_to_bhsd(qkvg, source_index, out, sequence, padded_sequence, 2, 128).chunk(4)
-    for name, old, new in zip(("q", "k", "v", "gate"), expected, actual):
+    for name, old, new in zip(("q", "k", "v", "gate"), expected, actual, strict=True):
         assert old.shape == new.shape, name
         assert torch.equal(old, new), name
+
+
+@pytest.mark.parametrize("fused", [False, True])
+def test_layout_benchmark_matches_forward_preprocessing(monkeypatch, fused):
+    if not torch.cuda.is_available():
+        pytest.skip("CUDA device required")
+
+    from fastvideo.attention.backends import video_sparse_attn as vsa
+
+    path = Path(__file__).resolve().parents[1] / "benchmarks/bench_vsa_tile_layout.py"
+    spec = importlib.util.spec_from_file_location("bench_vsa_tile_layout", path)
+    benchmark = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(benchmark)
+    original_gate = vsa._can_fuse_vsa64_layout
+
+    def allow_test_device(x, metadata):
+        with monkeypatch.context() as gate_patch:
+            gate_patch.setattr(torch.cuda, "get_device_capability", lambda *args: (10, 0))
+            return original_gate(x, metadata)
+
+    monkeypatch.setattr(vsa, "_can_fuse_vsa64_layout", allow_test_device)
+    impl = object.__new__(vsa.VideoSparseAttentionImpl)
+    builder = vsa.VideoSparseAttentionMetadataBuilder()
+    kwargs = dict(current_timestep=0,
+                  raw_latent_shape=(5, 5, 6),
+                  patch_size=(1, 1, 1),
+                  VSA_sparsity=0.8,
+                  device=torch.device("cuda"),
+                  cache_tile_buf=True)
+    qkvg = torch.randn((4, 157, 2, 128), device="cuda", dtype=torch.bfloat16)[:, :150]
+    captured = []
+
+    def capture_attention_inputs(q, k, v, *args, **kwargs):
+        captured[:] = [q, k, v, kwargs["compress_attn_weight"]]
+        return q
+
+    monkeypatch.setattr(vsa, "video_sparse_attn", capture_attention_inputs)
+    with torch.no_grad():
+        layout_step = benchmark.make_step(impl, qkvg, builder, kwargs, fused=fused, include_attn=False)
+        attention_step = benchmark.make_step(impl, qkvg, builder, kwargs, fused=fused, include_attn=True)
+
+        def reject_cat(*args, **kwargs):
+            pytest.fail("The timed layout step must not concatenate its outputs")
+
+        with monkeypatch.context() as timed_patch:
+            timed_patch.setattr(torch, "cat", reject_cat)
+            layout = layout_step()
+            attention = attention_step()
+
+    assert isinstance(layout, tuple) and len(layout) == 4
+    for actual, expected in zip(layout, captured, strict=True):
+        assert torch.equal(actual, expected)
+        assert actual.stride() == expected.stride()
+    assert all(t.is_contiguous() for t in layout[:3])
+    assert layout[3].is_contiguous() is fused
+    assert isinstance(attention, tuple) and len(attention) == 1
+    assert torch.equal(attention[0], captured[0].transpose(1, 2))

fastvideo/entrypoints/openai/media_policy.py

@@ -5,28 +5,23 @@ Local paths are readable only inside operator-listed ``server.media_roots``.
 URLs must use http(s) and, unless ``server.allow_private_media_urls`` is set,
 must resolve to public addresses on every redirect hop. Downloads are capped
 by streamed bytes and leave no partial file on failure or cancellation.
-Inline base64 ``data:`` URLs are decoded under the same byte cap.
 """
 
 from __future__ import annotations
 
 import asyncio
-import base64
-import binascii
 import contextlib
 import ipaddress
 import os
 import socket
 from dataclasses import dataclass
-from urllib.parse import unquote, urljoin, urlparse
+from urllib.parse import urljoin, urlparse
 
 import aiofiles
 import httpx
 
 MAX_REDIRECTS = 5
 READ_TIMEOUT_S = 60.0
-DATA_URL_HEADER_MAX_CHARS = 256
-DATA_URL_CHUNK_CHARS = 1 << 16  # at least 3, so a percent escape fits in one chunk
 
 
 class MediaPolicyError(ValueError):
@@ -79,73 +74,6 @@ def discard(path: str) -> None:
         os.unlink(path)
 
 
-def _too_large(max_bytes: int) -> MediaPolicyError:
-    return MediaPolicyError(f"Media reference is larger than the {max_bytes}-byte limit; raise "
-                            "server.media_reference_max_bytes in the serve config or provide a smaller file.")
-
-
-def data_url_media_type(source: str) -> str:
-    """The lower-cased media type of a base64 ``data:`` URL ("" when the URL names none)."""
-    separator = source.find(",")
-    if source[:5].lower() != "data:" or separator < 0 or separator > DATA_URL_HEADER_MAX_CHARS:
-        raise MediaPolicyError("Data URLs must have the form data:<media type>;base64,<payload>.")
-    parameters = source[5:separator].split(";")
-    if "base64" not in (parameter.strip().lower() for parameter in parameters[1:]):
-        raise MediaPolicyError("Data URLs must be base64-encoded: data:<media type>;base64,<payload>.")
-    return parameters[0].strip().lower()
-
-
-def _base64_payload(source: str, start: int, max_bytes: int) -> str:
-    """``source[start:]`` without percent escapes or whitespace, normalized in bounded chunks.
-
-    Stops as soon as the payload must decode to more than ``max_bytes``, so an oversized request costs memory
-    in proportion to the limit, not to its own size.
-    """
-    chunks: list[str] = []
-    length = 0
-    while start < len(source):
-        end = min(start + DATA_URL_CHUNK_CHARS, len(source))
-        escape = source.find("%", max(start + 1, end - 2), end) if end < len(source) else -1
-        if escape >= 0:
-            end = escape  # an escape that may continue past the chunk starts the next one
-        # Same leniency as SGLang Diffusion: percent escapes and line-wrapped payloads decode.
-        chunk = source[start:end]
-        if "%" in chunk:
-            chunk = unquote(chunk)
-        chunk = "".join(chunk.split())
-        length += len(chunk)
-        # Base64 decodes to 3 bytes per 4 characters, less at most 2 bytes of padding.
-        if length // 4 * 3 - 2 > max_bytes:
-            raise _too_large(max_bytes)
-        chunks.append(chunk)
-        start = end
-    return "".join(chunks)
-
-
-def decode_data_url(source: str, target: str, *, max_bytes: int) -> None:
-    """Decode a base64 ``data:`` URL to ``target``; nothing is written if it is invalid or over ``max_bytes``."""
-    data_url_media_type(source)
-    payload = _base64_payload(source, source.find(",") + 1, max_bytes)
-    padding = len(payload) - len(payload.rstrip("="))
-    if len(payload) // 4 * 3 - padding > max_bytes:
-        raise _too_large(max_bytes)
-    try:
-        data = base64.b64decode(payload, validate=True)
-    except (binascii.Error, ValueError) as error:
-        raise MediaPolicyError(f"Data URL payload is not valid base64: {error}") from error
-    if not data:
-        raise MediaPolicyError("Data URL payload is empty.")
-    if len(data) > max_bytes:
-        raise _too_large(max_bytes)
-    os.makedirs(os.path.dirname(os.path.abspath(target)), exist_ok=True)
-    try:
-        with open(target, "wb") as output:
-            output.write(data)
-    except BaseException:
-        discard(target)
-        raise
-
-
 async def fetch_media(
     url: str,
     target: str,
@@ -173,7 +101,9 @@ async def fetch_media(
                         async for chunk in response.aiter_bytes():
                             written += len(chunk)
                             if written > max_bytes:
-                                raise _too_large(max_bytes)
+                                raise MediaPolicyError(
+                                    f"Media reference is larger than the {max_bytes}-byte limit; raise "
+                                    "server.media_reference_max_bytes in the serve config or provide a smaller file.")
                             await output.write(chunk)
                     return response.headers
             raise MediaPolicyError(f"Media URL redirected more than {MAX_REDIRECTS} times.")

fastvideo/entrypoints/openai/request_adapter.py

@@ -12,8 +12,6 @@ from urllib.parse import urlparse
 from fastvideo.entrypoints.openai.media_policy import (
     MediaPolicy,
     MediaPolicyError,
-    data_url_media_type,
-    decode_data_url,
     discard,
     fetch_media,
     resolve_local_media,
@@ -43,23 +41,7 @@ class RequestAdaptationError(ValueError):
     """The transport request cannot be represented by the loaded pipeline."""
 
 
-_MEDIA_SUFFIXES = {
-    "video": frozenset({".avi", ".m4v", ".mkv", ".mov", ".mp4", ".mpeg", ".mpg", ".webm"}),
-    "audio": frozenset({".aac", ".flac", ".m4a", ".mp3", ".ogg", ".opus", ".wav", ".webm"}),
-}
-_DEFAULT_SUFFIXES = {"video": ".mp4", "audio": ".wav"}
-# Media types whose subtype is not already the file suffix; decoding probes content, so this is only a hint.
-_DATA_URL_SUFFIXES = {
-    "video/quicktime": ".mov",
-    "video/x-matroska": ".mkv",
-    "video/x-msvideo": ".avi",
-    "audio/mpeg": ".mp3",
-    "audio/mp4": ".m4a",
-    "audio/x-m4a": ".m4a",
-    "audio/wave": ".wav",
-    "audio/x-wav": ".wav",
-    "audio/x-flac": ".flac",
-}
+_VIDEO_SUFFIXES = frozenset({".avi", ".m4v", ".mkv", ".mov", ".mp4", ".mpeg", ".mpg", ".webm"})
 
 
 def _as_list(value: Any | list[Any] | None) -> list[Any]:
@@ -111,63 +93,30 @@ def _local_media(source: str, policy: MediaPolicy) -> str:
         raise RequestAdaptationError(str(error)) from error
 
 
-async def _decode_data_media(source: str, target_base: str, kind: str, *, max_bytes: int | None) -> str:
-    try:
-        media_type = data_url_media_type(source)
-        if media_type.partition("/")[0] != kind:
-            raise MediaPolicyError(f"{kind.capitalize()} references need a data:{kind}/... URL, got media type "
-                                   f"{media_type or 'none'!r}.")
-        subtype_suffix = f".{media_type.partition('/')[2]}"
-        suffix = _DATA_URL_SUFFIXES.get(media_type, subtype_suffix)
-        target = f"{target_base}{suffix if suffix in _MEDIA_SUFFIXES[kind] else _DEFAULT_SUFFIXES[kind]}"
-        decoding = asyncio.ensure_future(
-            asyncio.to_thread(decode_data_url,
-                              source,
-                              target,
-                              max_bytes=get_media_reference_max_bytes() if max_bytes is None else max_bytes))
-        try:
-            await asyncio.shield(decoding)
-        except asyncio.CancelledError:
-            # The decoding thread cannot be interrupted; remove its file once it has finished writing.
-            def remove(done: asyncio.Future) -> None:
-                if not done.cancelled():
-                    done.exception()  # mark it retrieved; the request is already gone
-                discard(target)
-
-            decoding.add_done_callback(remove)
-            raise
-    except MediaPolicyError as error:
-        raise RequestAdaptationError(str(error)) from error
-    return target
-
-
-async def _materialize_media(source: str,
+async def _materialize_video(source: str,
                              target_base: str,
-                             kind: str,
                              *,
                              max_bytes: int | None = None,
-                             policy: MediaPolicy | None = None) -> tuple[str, bool]:
-    """Return a local file for a video or audio reference and whether this request created it."""
+                             policy: MediaPolicy | None = None) -> str:
     policy = get_media_policy() if policy is None else policy
     source = source.strip()
     if not source:
-        raise RequestAdaptationError(f"{kind.capitalize()} references must not be empty.")
-    # Only the scheme is lower-cased: copying a whole data: URL would double its memory.
-    if source[:8].lower().startswith(("http://", "https://")):
+        raise RequestAdaptationError("Video references must not be empty.")
+    if source.lower().startswith(("http://", "https://")):
         suffix = os.path.splitext(urlparse(source).path)[1].lower()
-        if suffix not in _MEDIA_SUFFIXES[kind]:
-            suffix = _DEFAULT_SUFFIXES[kind]
+        if suffix not in _VIDEO_SUFFIXES:
+            suffix = ".mp4"
         target = f"{target_base}{suffix}"
         try:
             await _download_media(source, target, max_bytes=max_bytes, policy=policy)
         except RequestAdaptationError:
             raise
         except Exception as error:
-            raise RequestAdaptationError(f"Unable to fetch {kind} reference: {error}") from error
-        return target, True
-    if source[:5].lower() == "data:":
-        return await _decode_data_media(source, target_base, kind, max_bytes=max_bytes), True
-    return _local_media(source, policy), False
+            raise RequestAdaptationError(f"Unable to fetch video reference: {error}") from error
+        return target
+    if source.lower().startswith("data:"):
+        raise RequestAdaptationError("Data video references are not supported; provide a URL or local path.")
+    return _local_media(source, policy)
 
 
 async def _materialize_image_url(source: str, target_base: str, *, max_bytes: int | None, policy: MediaPolicy) -> str:
@@ -214,8 +163,6 @@ async def prepare_reference_media(
                 local_path = track(await save_base64_image_to_path(source, target))
             except Exception as error:
                 raise RequestAdaptationError(f"Unable to decode image reference: {error}") from error
-        elif source[:5].lower() == "data:":
-            raise RequestAdaptationError("Image references need a data:image/... URL.")
         elif is_remote(source):
             local_path = track(await _materialize_image_url(source, target, max_bytes=max_bytes, policy=policy))
         else:
@@ -226,14 +173,6 @@ async def prepare_reference_media(
             raise RequestAdaptationError(f"Unable to decode image reference: {error}") from error
         return local_path
 
-    async def materialize_clip(source: str, name: str, kind: str) -> str:
-        local_path, created_here = await _materialize_media(source,
-                                                            os.path.join(uploads_dir, f"{request_id}_{name}"),
-                                                            kind,
-                                                            max_bytes=max_bytes,
-                                                            policy=policy)
-        return track(local_path) if created_here else local_path
-
     try:
         index = 0
         for reference in _as_list(request.image_reference):
@@ -252,13 +191,29 @@ async def prepare_reference_media(
 
         for index, reference in enumerate(_as_list(request.video_reference)):
             if isinstance(reference, UrlVideoReference):
-                reference.video_url = await materialize_clip(reference.video_url, f"video_{index}", "video")
+                remote = is_remote(reference.video_url)
+                reference.video_url = await _materialize_video(reference.video_url,
+                                                               os.path.join(uploads_dir, f"{request_id}_video_{index}"),
+                                                               max_bytes=max_bytes,
+                                                               policy=policy)
+                if remote:
+                    track(reference.video_url)
         if request.video_path:
-            request.video_path = await materialize_clip(request.video_path, "video_path", "video")
+            remote = is_remote(request.video_path)
+            request.video_path = await _materialize_video(request.video_path,
+                                                          os.path.join(uploads_dir, f"{request_id}_video_path"),
+                                                          max_bytes=max_bytes,
+                                                          policy=policy)
+            if remote:
+                track(request.video_path)
         if request.video_url:
-            request.video_url = await materialize_clip(request.video_url, "video_url", "video")
-        for index, reference in enumerate(_as_list(request.audio_reference)):
-            reference.audio_url = await materialize_clip(reference.audio_url, f"audio_{index}", "audio")
+            remote = is_remote(request.video_url)
+            request.video_url = await _materialize_video(request.video_url,
+                                                         os.path.join(uploads_dir, f"{request_id}_video_url"),
+                                                         max_bytes=max_bytes,
+                                                         policy=policy)
+            if remote:
+                track(request.video_url)
     except BaseException:
         # Includes cancellation, so a disconnected client leaves no downloaded inputs behind.
         discard_materialized_media(created)

fastvideo/layers/lora/linear.py

@@ -155,37 +155,43 @@ class BaseLayerWithLoRA(nn.Module):
         """The wrapped layer's bias, or ``None`` when it has none."""
         return getattr(self.base_layer, "bias", None)
 
-    @torch.compile()
     def forward(self, x: torch.Tensor) -> torch.Tensor:
-        lora_A = self.lora_A
-        lora_B = self.lora_B
-        if isinstance(self.lora_B, DTensor):
-            lora_B = self.lora_B.to_local()
-            lora_A = self.lora_A.to_local()
-
-        if self.side_branch and (self.lora_A is None or self._strength_is_zero):
-            # No adapter, or strength 0, is exactly the unadapted layer, down to the output dtype.
+        # The wrapper adds no compiled region around the base layer, so it runs exactly as it would unwrapped: eagerly,
+        # or inside a caller's compiled block. Compiling a packed NVFP4/FP8 linear on its own changes its numerics, and
+        # a wrapped layer with no adapter or strength 0 must equal the unwrapped layer exactly.
+        if self.lora_A is None or (self.side_branch and self._strength_is_zero):
+            # No adapter, or strength 0 on a side branch, is exactly the unadapted layer, down to the output dtype.
+            # A merge-mode wrapper reaches this branch while its first adapter is still loading or after a load
+            # failed: the pipeline wraps every linear before it reads the weights, so an unadapted request must
+            # not depend on an adapter that is not there.
             return self.base_layer(x)
 
-        if not self.merged and not self.disable_lora:
-            lora_A_sliced = self.slice_lora_a_weights(lora_A.to(x, non_blocking=True))
-            lora_B_sliced = self.slice_lora_b_weights(lora_B.to(x, non_blocking=True))
-            alpha_over_rank = 1.0
-            if self.lora_alpha != self.lora_rank:
-                alpha_over_rank = float(self.lora_alpha) / float(self.lora_rank)  # type: ignore[arg-type]
-            out, output_bias = self.base_layer(x)
-            if self.side_branch:
-                # Keep chunk temporaries opaque to Inductor so fullgraph capture cannot retain a full-size delta.
-                _add_low_rank_inplace(out, x, lora_A_sliced, lora_B_sliced, self._side_branch_strength, alpha_over_rank)
-                return out, output_bias
-            scale = float(self.lora_strength) * alpha_over_rank
-            delta = x @ lora_A_sliced.transpose(0, 1) @ lora_B_sliced.transpose(0, 1)
-            if scale != 1.0:
-                delta = delta * scale
-            return out + delta, output_bias
-        else:
-            out, output_bias = self.base_layer(x)
+        out, output_bias = self.base_layer(x)
+        if self.merged or self.disable_lora:
             return out.to(x), output_bias
+        return self._add_adapter(out, x), output_bias
+
+    @torch.compile()
+    def _add_adapter(self, out: torch.Tensor, x: torch.Tensor) -> torch.Tensor:
+        lora_A = self.lora_A
+        lora_B = self.lora_B
+        if isinstance(lora_B, DTensor):
+            lora_B = lora_B.to_local()
+            lora_A = lora_A.to_local()
+        lora_A_sliced = self.slice_lora_a_weights(lora_A.to(x, non_blocking=True))
+        lora_B_sliced = self.slice_lora_b_weights(lora_B.to(x, non_blocking=True))
+        alpha_over_rank = 1.0
+        if self.lora_alpha != self.lora_rank:
+            alpha_over_rank = float(self.lora_alpha) / float(self.lora_rank)  # type: ignore[arg-type]
+        if self.side_branch:
+            # Keep chunk temporaries opaque to Inductor so fullgraph capture cannot retain a full-size delta.
+            _add_low_rank_inplace(out, x, lora_A_sliced, lora_B_sliced, self._side_branch_strength, alpha_over_rank)
+            return out
+        scale = float(self.lora_strength) * alpha_over_rank
+        delta = x @ lora_A_sliced.transpose(0, 1) @ lora_B_sliced.transpose(0, 1)
+        if scale != 1.0:
+            delta = delta * scale
+        return out + delta
 
     def slice_lora_a_weights(self, A: torch.Tensor) -> torch.Tensor:
         return A

fastvideo/tests/entrypoints/test_openai_serving_engine.py

@@ -4,8 +4,6 @@
 from __future__ import annotations
 
 import asyncio
-import base64
-import os
 from pathlib import Path
 import threading
 import time
@@ -17,13 +15,11 @@ from fastapi.testclient import TestClient
 
 from fastvideo.api.parser import parse_config
 from fastvideo.api.schema import GenerationRequest
-from fastvideo.entrypoints.openai import media_policy, request_adapter
 from fastvideo.entrypoints.openai.media_policy import MediaPolicy
-from fastvideo.entrypoints.openai.protocol import UrlAudioReference, VideoGenerationRequest
+from fastvideo.entrypoints.openai.protocol import VideoGenerationRequest
 from fastvideo.entrypoints.openai.request_adapter import (
     RequestAdaptationError,
     build_generation_request,
-    discard_materialized_media,
     prepare_reference_media,
     validate_served_model_name,
 )
@@ -245,8 +241,7 @@ def test_wav_media_type_is_detected_from_output_path() -> None:
 def test_missing_video_reference_fails_during_admission(tmp_path: Path) -> None:
     request = VideoGenerationRequest(prompt="waves", video_reference={"video_url": str(tmp_path / "missing.mp4")})
     with pytest.raises(RequestAdaptationError, match="does not exist"):
-        asyncio.run(
-            prepare_reference_media("request", request, str(tmp_path), policy=MediaPolicy(roots=(str(tmp_path), ))))
+        asyncio.run(prepare_reference_media("request", request, str(tmp_path), policy=MediaPolicy(roots=(str(tmp_path), ))))
 
 
 def test_local_media_paths_need_a_configured_root(tmp_path: Path) -> None:
@@ -257,166 +252,6 @@ def test_local_media_paths_need_a_configured_root(tmp_path: Path) -> None:
         asyncio.run(prepare_reference_media("request", request, str(tmp_path), policy=MediaPolicy()))
 
 
-def _data_url(media_type: str, payload: bytes) -> str:
-    return f"data:{media_type};base64,{base64.b64encode(payload).decode()}"
-
-
-def _ref(kind: str, url: str) -> dict[str, str]:
-    return {f"{kind}_url": url}
-
-
-def _uploads(tmp_path: Path) -> list[Path]:
-    return list((tmp_path / "uploads").glob("*"))
-
-
-def test_inline_video_and_audio_references_are_decoded_in_order(tmp_path: Path) -> None:
-    request = VideoGenerationRequest(
-        prompt="waves",
-        video_reference=[
-            _ref("video", _data_url("video/mp4", b"first-clip")),
-            _ref("video", _data_url("video/quicktime", b"second-clip")),
-        ],
-        audio_reference=[
-            _ref("audio", _data_url("audio/mpeg", b"first-track")),
-            _ref("audio", _data_url("audio/wav", b"second-track")),
-        ],
-    )
-    created = asyncio.run(prepare_reference_media("request", request, str(tmp_path), policy=MediaPolicy()))
-    assert isinstance(request.video_reference, list) and isinstance(request.audio_reference, list)
-    videos = [reference.video_url for reference in request.video_reference]
-    audios = [reference.audio_url for reference in request.audio_reference]
-    assert created == videos + audios
-    contents = [Path(path).read_bytes() for path in created]
-    assert contents == [b"first-clip", b"second-clip", b"first-track", b"second-track"]
-    assert [Path(path).suffix for path in created] == [".mp4", ".mov", ".mp3", ".wav"]
-    assert {Path(path).parent for path in created} == {tmp_path / "uploads"}
-    discard_materialized_media(created)
-    assert not _uploads(tmp_path)
-
-
-@pytest.mark.parametrize("chunk_chars", [media_policy.DATA_URL_CHUNK_CHARS, 3, 4, 5, 7])
-def test_inline_reference_payload_may_be_wrapped_and_percent_escaped(tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
-                                                                    chunk_chars: int) -> None:
-    # Small chunks put chunk boundaries inside percent escapes and line breaks.
-    monkeypatch.setattr(media_policy, "DATA_URL_CHUNK_CHARS", chunk_chars)
-    track = bytes(range(256)) * 2
-    encoded = base64.b64encode(track).decode()
-    lines = [encoded[start:start + 16] for start in range(0, len(encoded), 16)]
-    wrapped = "".join(line + ("%0A" if number % 2 else "\r\n") for number, line in enumerate(lines))
-    escaped = wrapped.replace("+", "%2B").replace("/", "%2f").replace("=", "%3D")
-    request = VideoGenerationRequest(prompt="waves", audio_reference={"audio_url": f"data:audio/flac;base64,{escaped}"})
-    created = asyncio.run(prepare_reference_media("request", request, str(tmp_path), policy=MediaPolicy()))
-    assert Path(created[0]).read_bytes() == track
-    assert created[0].endswith(".flac")
-
-
-@pytest.mark.parametrize(
-    ("kind", "url", "message"),
-    [
-        ("video", "data:video/mp4;base64,not*base64", "not valid base64"),
-        ("audio", "data:audio/mpeg;base64,", "payload is empty"),
-        ("video", "data:video/mp4,raw-bytes", "must be base64-encoded"),
-        ("video", _data_url("audio/mpeg", b"track"), "need a data:video/"),
-        ("audio", _data_url("video/mp4", b"clip"), "need a data:audio/"),
-        ("audio", _data_url("application/octet-stream", b"track"), "need a data:audio/"),
-        ("image", _data_url("video/mp4", b"clip"), "need a data:image/"),
-    ],
-)
-def test_invalid_inline_references_are_refused(tmp_path: Path, kind: str, url: str, message: str) -> None:
-    request = VideoGenerationRequest(prompt="waves", **{f"{kind}_reference": _ref(kind, url)})
-    with pytest.raises(RequestAdaptationError, match=message):
-        asyncio.run(prepare_reference_media("request", request, str(tmp_path), policy=MediaPolicy()))
-    assert not _uploads(tmp_path)
-
-
-def test_inline_reference_over_the_media_limit_is_refused(tmp_path: Path) -> None:
-    request = VideoGenerationRequest(prompt="waves", audio_reference={"audio_url": _data_url("audio/wav", b"12345")})
-    with pytest.raises(RequestAdaptationError, match="4-byte limit"):
-        asyncio.run(prepare_reference_media("request", request, str(tmp_path), max_bytes=4, policy=MediaPolicy()))
-    assert not _uploads(tmp_path)
-    at_limit = VideoGenerationRequest(prompt="waves", video_reference={"video_url": _data_url("video/mp4", b"1234")})
-    created = asyncio.run(prepare_reference_media("request", at_limit, str(tmp_path), max_bytes=4,
-                                                  policy=MediaPolicy()))
-    assert Path(created[0]).read_bytes() == b"1234"
-
-
-def test_oversized_inline_payload_is_refused_before_it_is_all_read(tmp_path: Path,
-                                                                   monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setattr(media_policy, "DATA_URL_CHUNK_CHARS", 8)
-    unescaped: list[int] = []
-    original_unquote = media_policy.unquote
-
-    def counting_unquote(chunk: str) -> str:
-        unescaped.append(len(chunk))
-        return original_unquote(chunk)
-
-    monkeypatch.setattr(media_policy, "unquote", counting_unquote)
-    source = "data:video/mp4;base64," + "QU%4AB\n" * 10_000
-    with pytest.raises(media_policy.MediaPolicyError, match="4-byte limit"):
-        media_policy.decode_data_url(source, str(tmp_path / "clip.mp4"), max_bytes=4)
-    assert max(unescaped) <= 8 and sum(unescaped) < 64
-    assert not (tmp_path / "clip.mp4").exists()
-
-
-def test_decoded_references_are_removed_when_a_later_reference_fails(tmp_path: Path) -> None:
-    request = VideoGenerationRequest(
-        prompt="waves",
-        video_reference={"video_url": _data_url("video/mp4", b"clip")},
-        audio_reference={"audio_url": "data:audio/mpeg;base64,%%%"},
-    )
-    with pytest.raises(RequestAdaptationError, match="not valid base64"):
-        asyncio.run(prepare_reference_media("request", request, str(tmp_path), policy=MediaPolicy()))
-    assert not _uploads(tmp_path)
-
-
-def test_cancelled_decoding_removes_its_file_once_the_thread_finishes(tmp_path: Path,
-                                                                      monkeypatch: pytest.MonkeyPatch) -> None:
-    asyncio.run(_assert_cancelled_decoding_removes_its_file(tmp_path, monkeypatch))
-
-
-async def _assert_cancelled_decoding_removes_its_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
-    started, release, finished = threading.Event(), threading.Event(), threading.Event()
-    written: list[bool] = []
-
-    def decode_after_release(source: str, target: str, *, max_bytes: int) -> None:
-        started.set()
-        release.wait(timeout=5)
-        try:
-            media_policy.decode_data_url(source, target, max_bytes=max_bytes)
-            written.append(os.path.exists(target))
-        finally:
-            finished.set()
-
-    monkeypatch.setattr(request_adapter, "decode_data_url", decode_after_release)
-    request = VideoGenerationRequest(prompt="waves", video_reference={"video_url": _data_url("video/mp4", b"clip")})
-    preparing = asyncio.create_task(prepare_reference_media("request", request, str(tmp_path), policy=MediaPolicy()))
-    assert await asyncio.to_thread(started.wait, 5)
-    preparing.cancel()
-    with pytest.raises(asyncio.CancelledError):
-        await preparing
-    # The thread writes its file only after the request is gone, so cleanup has to wait for the thread.
-    release.set()
-    assert await asyncio.to_thread(finished.wait, 5)
-    assert written == [True]
-    deadline = time.monotonic() + 5
-    while _uploads(tmp_path) and time.monotonic() < deadline:
-        await asyncio.sleep(0.01)
-    assert not _uploads(tmp_path)
-
-
-def test_local_audio_paths_need_a_configured_root(tmp_path: Path) -> None:
-    track = tmp_path / "track.wav"
-    track.write_bytes(b"x")
-    request = VideoGenerationRequest(prompt="waves", audio_reference={"audio_url": str(track)})
-    with pytest.raises(RequestAdaptationError, match="server.media_roots"):
-        asyncio.run(prepare_reference_media("request", request, str(tmp_path), policy=MediaPolicy()))
-    created = asyncio.run(
-        prepare_reference_media("request", request, str(tmp_path), policy=MediaPolicy(roots=(str(tmp_path), ))))
-    assert created == []
-    assert isinstance(request.audio_reference, UrlAudioReference)
-    assert request.audio_reference.audio_url == os.path.realpath(track)
-
-
 def test_mmaudio_v2a_requires_video_reference(tmp_path: Path) -> None:
     request = VideoGenerationRequest(prompt="waves")
     with pytest.raises(RequestAdaptationError, match="requires video_reference"):
@@ -432,18 +267,13 @@ def test_mmaudio_v2a_requires_video_reference(tmp_path: Path) -> None:
 def test_missing_reference_fails_during_admission(tmp_path: Path) -> None:
     request = VideoGenerationRequest(prompt="a fox", input_reference=str(tmp_path / "missing.png"))
     with pytest.raises(RequestAdaptationError, match="does not exist"):
-        asyncio.run(
-            prepare_reference_media("request", request, str(tmp_path), policy=MediaPolicy(roots=(str(tmp_path), ))))
+        asyncio.run(prepare_reference_media("request", request, str(tmp_path), policy=MediaPolicy(roots=(str(tmp_path), ))))
 
 
 def test_request_adapter_resolves_vllm_nested_params(tmp_path: Path) -> None:
     request = VideoGenerationRequest(
         prompt="a fox",
-        video_params={
-            "width": 832,
-            "height": 480,
-            "fps": 30
-        },
+        video_params={"width": 832, "height": 480, "fps": 30},
         seconds="2",
         seed=7,
     )
@@ -474,11 +304,7 @@ def test_request_adapter_accepts_matching_startup_lora(tmp_path: Path) -> None:
     request = VideoGenerationRequest(
         prompt="a fox",
         model="fast",
-        lora={
-            "name": "fast",
-            "path": adapter,
-            "scale": 0.75
-        },
+        lora={"name": "fast", "path": adapter, "scale": 0.75},
     )
 
     build_generation_request(
@@ -533,22 +359,10 @@ def test_fasth3_request_uses_the_general_adapter(tmp_path: Path) -> None:
 @pytest.mark.parametrize(
     ("overrides", "message"),
     [
-        ({
-            "aspect_ratio": "16:9",
-            "num_frames": 22
-        }, "generates 5-15 seconds"),
-        ({
-            "size": "100x100",
-            "num_frames": 124
-        }, "positive multiples of 32"),
-        ({
-            "aspect_ratio": "1:9",
-            "num_frames": 124
-        }, "supports aspect ratios"),
-        ({
-            "aspect_ratio": "16:9",
-            "num_frames": 120
-        }, "causal-VAE grid"),
+        ({"aspect_ratio": "16:9", "num_frames": 22}, "generates 5-15 seconds"),
+        ({"size": "100x100", "num_frames": 124}, "positive multiples of 32"),
+        ({"aspect_ratio": "1:9", "num_frames": 124}, "supports aspect ratios"),
+        ({"aspect_ratio": "16:9", "num_frames": 120}, "causal-VAE grid"),
     ],
 )
 def test_fasth3_invalid_geometry_fails_at_admission(tmp_path: Path, overrides, message) -> None:
@@ -582,9 +396,7 @@ def test_fasth3_ref2va_limits_references_at_admission(tmp_path: Path) -> None:
         task="ref2va",
         aspect_ratio="16:9",
         num_frames=124,
-        image_reference=[{
-            "image_url": f"/tmp/reference-{index}.png"
-        } for index in range(10)],
+        image_reference=[{"image_url": f"/tmp/reference-{index}.png"} for index in range(10)],
     )
     with pytest.raises(RequestAdaptationError, match="at most 9 image references"):
         build_generation_request(
@@ -677,11 +489,7 @@ def test_audio_jobs_publish_wav_media_type_and_seconds(tmp_path: Path) -> None:
         with TestClient(app) as client:
             created = client.post(
                 "/v1/videos",
-                json={
-                    "model": "stable-audio",
-                    "prompt": "rain",
-                    "seconds": 3
-                },
+                json={"model": "stable-audio", "prompt": "rain", "seconds": 3},
             )
             assert created.status_code == 200
             audio_id = created.json()["id"]
@@ -714,69 +522,6 @@ def test_audio_jobs_publish_wav_media_type_and_seconds(tmp_path: Path) -> None:
         state.clear_state()
 
 
-class _ReferenceGenerator(_FileGenerator):
-
-    def __init__(self) -> None:
-        super().__init__()
-        self.references: list[tuple[str, bytes]] = []
-
-    def generate(self, request):
-        self.references = [(reference.media_type, Path(reference.source).read_bytes())
-                           for reference in request.inputs.references]
-        return super().generate(request)
-
-
-def test_ref2va_route_accepts_inline_references_and_removes_them(tmp_path: Path) -> None:
-    from fastvideo.entrypoints.openai import state
-    from fastvideo.entrypoints.openai.video_api import router
-
-    generator = _ReferenceGenerator()
-    engine = OpenAIServingEngine(generator)  # type: ignore[arg-type]
-    args = _args("MiniMaxAI/MiniMax-H3", override_pipeline_cls_name="MiniMaxH3Ref2VAModularPipeline")
-    state.set_state(
-        generator,  # type: ignore[arg-type]
-        engine,
-        args,  # type: ignore[arg-type]
-        str(tmp_path),
-        served_model_name="minimax-h3",
-    )
-    asyncio.run(VIDEO_STORE.clear())
-    app = FastAPI()
-    app.include_router(router)
-    body = dict(
-        model="minimax-h3",
-        prompt="a fox sings along",
-        task="ref2va",
-        aspect_ratio="16:9",
-        num_frames=124,
-        video_reference=[_ref("video", _data_url("video/mp4", b"clip-with-sound"))],
-        audio_reference=[
-            _ref("audio", _data_url("audio/mpeg", b"first-voice")),
-            _ref("audio", _data_url("audio/wav", b"second-voice")),
-        ],
-    )
-
-    try:
-        with TestClient(app) as client:
-            sync = client.post("/v1/videos/sync", json=body)
-            assert sync.status_code == 200, sync.text
-            assert generator.references == [
-                ("video", b"clip-with-sound"),
-                ("audio", b"first-voice"),
-                ("audio", b"second-voice"),
-            ]
-            assert not _uploads(tmp_path)
-
-            mismatched = {**body, "audio_reference": [_ref("audio", _data_url("video/mp4", b"clip"))]}
-            refused = client.post("/v1/videos", json=mismatched)
-            assert refused.status_code == 400
-            assert "need a data:audio/" in refused.text
-            assert not _uploads(tmp_path)
-    finally:
-        asyncio.run(VIDEO_STORE.clear())
-        state.clear_state()
-
-
 def test_video_routes_cover_async_sync_list_content_and_delete(tmp_path: Path) -> None:
     from fastvideo.entrypoints.openai import state
     from fastvideo.entrypoints.openai.video_api import router

fastvideo/tests/layers/test_lora_side_branch.py

@@ -54,6 +54,89 @@ def _packed_linear(seed: int = 0) -> ReplicatedLinear:
     return layer
 
 
+def test_a_wrapped_layer_without_an_adapter_runs_the_base_layer() -> None:
+    """A dense wrapper with no adapter is the base layer.
+
+    The pipeline wraps every linear when the first runtime adapter is set. A render that then runs without one, or a
+    load that failed part-way, must still produce the base model's output instead of failing on the missing factors.
+    """
+    base = _dense_linear()
+    wrapped = get_lora_layer(base)
+    assert not wrapped.side_branch and wrapped.lora_A is None
+    x = torch.randn(2, IN, dtype=torch.bfloat16)
+    with torch.no_grad():
+        assert torch.equal(wrapped(x)[0], base(x)[0])
+
+
+@pytest.mark.parametrize("packed", [False, True])
+def test_wrapped_layer_at_strength_zero_is_bitwise_the_base_layer(packed: bool) -> None:
+    base = _packed_linear() if packed else _dense_linear()
+    x = _x()
+    with torch.no_grad():
+        # Taken before wrapping, so a merge that wrongly rewrites the shared base weight cannot also change it.
+        expected = base(x)[0].clone()
+    layer = get_lora_layer(base)
+    layer.set_lora_weights(*_factors(), lora_alpha=8, strength=0.0)
+    assert layer.lora_A is not None and layer.lora_strength == 0.0
+    with torch.no_grad():
+        assert torch.equal(layer(x)[0], expected)
+
+
+@pytest.mark.parametrize("backend", ["eager", "inductor"])
+@pytest.mark.parametrize("adapter", [False, True])
+def test_wrapping_does_not_change_a_compiled_block(monkeypatch: pytest.MonkeyPatch, backend: str,
+                                                   adapter: bool) -> None:
+    """Under a caller's compile the base layer is traced with the block, as it is unwrapped; the wrapper adds no
+    region of its own, so the compiled block is unchanged with no adapter or at strength 0."""
+    if backend == "inductor" and not torch.cuda.is_available():
+        pytest.skip("Inductor execution requires a CUDA GPU for this regression")
+    device = torch.device("cuda" if backend == "inductor" else "cpu")
+    monkeypatch.setattr(torch._dynamo.config, "disable", False)
+    torch._dynamo.reset()
+
+    class Block(torch.nn.Module):
+
+        def __init__(self, proj: torch.nn.Module) -> None:
+            super().__init__()
+            self.proj = proj
+
+        def forward(self, x: torch.Tensor) -> torch.Tensor:
+            return self.proj(x)[0] + x[..., :OUT]
+
+    x = _x().to(device)
+    try:
+        with torch.no_grad():
+            expected = torch.compile(Block(_packed_linear().to(device)), backend=backend, fullgraph=True)(x)
+            torch._dynamo.reset()
+            layer = get_lora_layer(_packed_linear().to(device))
+            if adapter:
+                layer.set_lora_weights(*(factor.to(device) for factor in _factors()), lora_alpha=8, strength=0.0)
+            actual = torch.compile(Block(layer), backend=backend, fullgraph=True)(x)
+        assert torch.equal(actual, expected)
+    finally:
+        torch._dynamo.reset()
+
+
+def test_wrapper_adds_no_compiled_region_around_the_base_layer(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Compiling a packed base layer on its own changes its numerics, so an uncompiled caller runs it eagerly."""
+    monkeypatch.setattr(torch._dynamo.config, "disable", False)
+    torch._dynamo.reset()
+    base = _packed_linear()
+    layer = get_lora_layer(base)
+    layer.set_lora_weights(*_factors(), lora_alpha=8, strength=0.5)
+    seen: list[bool] = []
+    original = base.forward
+
+    def spy(x: torch.Tensor):
+        seen.append(torch.compiler.is_compiling())
+        return original(x)
+
+    monkeypatch.setattr(base, "forward", spy)
+    with torch.no_grad():
+        layer(_x())
+    assert seen == [False]
+
+
 def _factors(seed: int = 1) -> tuple[torch.Tensor, torch.Tensor]:
     generator = torch.Generator().manual_seed(seed)
     return torch.randn(RANK, IN, generator=generator), torch.randn(OUT, RANK, generator=generator)