[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/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/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