[feat] OpenAI data URLs: decode in bounded chunks, test cancelled decoding

Review follow-ups for the inline video/audio reference decoder:

- decode_data_url now undoes percent escapes and strips whitespace in
  64 KiB chunks and stops as soon as the payload must decode to more than
  server.media_reference_max_bytes. An oversized request costs memory in
  proportion to the limit, not to its own size. Escapes are never split
  across chunks.
- Data URL checks lower-case only the scheme instead of copying the whole
  URL.
- Tests: the wrapped/percent-escaped payload decodes the same at chunk
  sizes that split escapes and line breaks; an oversized payload is refused
  after a few chunks; and a request cancelled before its decoding thread
  writes the file still leaves no upload once the thread finishes.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

docs/design/server_contracts/openai.md

@@ -88,12 +88,27 @@ 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 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.
+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.
 
 ## 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 + three contiguous Q/K/V transposes + a gate view
+  - legacy: tile scatter into BSHD + four BSHD->BHSD transposes
   - 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[[], tuple[torch.Tensor, ...]]:
+              build_kwargs: dict, fused: bool, include_attn: bool) -> Callable[[], torch.Tensor]:
     metadata = builder.build(**build_kwargs)
 
-    def step() -> tuple[torch.Tensor, ...]:
+    def step() -> 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 q, k, v, gate
-        return (q.transpose(1, 2).contiguous(), k.transpose(1, 2).contiguous(), v.transpose(1, 2).contiguous(),
-                gate.transpose(1, 2))
+            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)])
 
     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 all(torch.equal(old, new) for old, new in zip(legacy(), fused(), strict=True)):
+        if not torch.equal(legacy(), fused()):
             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,9 +5,6 @@ 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
 
@@ -43,63 +40,6 @@ 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, strict=True):
+    for name, old, new in zip(("q", "k", "v", "gate"), expected, actual):
         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,23 +5,28 @@ 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 urljoin, urlparse
+from urllib.parse import unquote, 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):
@@ -74,6 +79,73 @@ 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,
@@ -101,9 +173,7 @@ async def fetch_media(
                         async for chunk in response.aiter_bytes():
                             written += len(chunk)
                             if written > 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.")
+                                raise _too_large(max_bytes)
                             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,6 +12,8 @@ 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,
@@ -41,7 +43,23 @@ class RequestAdaptationError(ValueError):
     """The transport request cannot be represented by the loaded pipeline."""
 
 
-_VIDEO_SUFFIXES = frozenset({".avi", ".m4v", ".mkv", ".mov", ".mp4", ".mpeg", ".mpg", ".webm"})
+_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",
+}
 
 
 def _as_list(value: Any | list[Any] | None) -> list[Any]:
@@ -93,30 +111,63 @@ def _local_media(source: str, policy: MediaPolicy) -> str:
         raise RequestAdaptationError(str(error)) from error
 
 
-async def _materialize_video(source: str,
+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,
                              target_base: str,
+                             kind: str,
                              *,
                              max_bytes: int | None = None,
-                             policy: MediaPolicy | None = None) -> str:
+                             policy: MediaPolicy | None = None) -> tuple[str, bool]:
+    """Return a local file for a video or audio reference and whether this request created it."""
     policy = get_media_policy() if policy is None else policy
     source = source.strip()
     if not source:
-        raise RequestAdaptationError("Video references must not be empty.")
-    if source.lower().startswith(("http://", "https://")):
+        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://")):
         suffix = os.path.splitext(urlparse(source).path)[1].lower()
-        if suffix not in _VIDEO_SUFFIXES:
-            suffix = ".mp4"
+        if suffix not in _MEDIA_SUFFIXES[kind]:
+            suffix = _DEFAULT_SUFFIXES[kind]
         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 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)
+            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
 
 
 async def _materialize_image_url(source: str, target_base: str, *, max_bytes: int | None, policy: MediaPolicy) -> str:
@@ -163,6 +214,8 @@ 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:
@@ -173,6 +226,14 @@ 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):
@@ -191,29 +252,13 @@ async def prepare_reference_media(
 
         for index, reference in enumerate(_as_list(request.video_reference)):
             if isinstance(reference, UrlVideoReference):
-                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)
+                reference.video_url = await materialize_clip(reference.video_url, f"video_{index}", "video")
         if request.video_path:
-            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)
+            request.video_path = await materialize_clip(request.video_path, "video_path", "video")
         if request.video_url:
-            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)
+            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")
     except BaseException:
         # Includes cancellation, so a disconnected client leaves no downloaded inputs behind.
         discard_materialized_media(created)

fastvideo/layers/lora/linear.py

@@ -155,43 +155,37 @@ class BaseLayerWithLoRA(nn.Module):
         """The wrapped layer's bias, or ``None`` when it has none."""
         return getattr(self.base_layer, "bias", None)
 
-    def forward(self, x: torch.Tensor) -> torch.Tensor:
-        # 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)
-
-        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:
+    def forward(self, 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
+        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.
+            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)
+            return out.to(x), output_bias
 
     def slice_lora_a_weights(self, A: torch.Tensor) -> torch.Tensor:
         return A

fastvideo/tests/entrypoints/test_openai_serving_engine.py

@@ -4,6 +4,8 @@
 from __future__ import annotations
 
 import asyncio
+import base64
+import os
 from pathlib import Path
 import threading
 import time
@@ -15,11 +17,13 @@ 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 VideoGenerationRequest
+from fastvideo.entrypoints.openai.protocol import UrlAudioReference, VideoGenerationRequest
 from fastvideo.entrypoints.openai.request_adapter import (
     RequestAdaptationError,
     build_generation_request,
+    discard_materialized_media,
     prepare_reference_media,
     validate_served_model_name,
 )
@@ -241,7 +245,8 @@ 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:
@@ -252,6 +257,166 @@ 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"):
@@ -267,13 +432,18 @@ 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,
     )
@@ -304,7 +474,11 @@ 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(
@@ -359,10 +533,22 @@ 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:
@@ -396,7 +582,9 @@ 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(
@@ -489,7 +677,11 @@ 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"]
@@ -522,6 +714,69 @@ 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,89 +54,6 @@ 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)