diff --git a/docs/source/commands/trtllm-serve/trtllm-serve.rst b/docs/source/commands/trtllm-serve/trtllm-serve.rst index 5f9297eab256..4f0312fe139c 100644 --- a/docs/source/commands/trtllm-serve/trtllm-serve.rst +++ b/docs/source/commands/trtllm-serve/trtllm-serve.rst @@ -236,7 +236,7 @@ Visual Generation Serving trtllm-serve nvidia/Cosmos3-Nano \ --enable_visual_gen -For checkpoints that support both LLM and Visual Generation, such as Cosmos3, pass ``--enable_visual_gen`` to select the VisualGen runtime when ``--visual_gen_args`` is not specified. The ``--visual_gen_args`` flag accepts a YAML file that configures quantization, parallelism, and TeaCache. Available visual generation endpoints include ``/v1/images/generations``, ``/v1/videos``, ``/v1/videos/generations``, and video management APIs. +For checkpoints that support both LLM and Visual Generation, such as Cosmos3, pass ``--enable_visual_gen`` to select the VisualGen runtime when ``--visual_gen_args`` is not specified. The ``--visual_gen_args`` flag accepts a YAML file that configures quantization, parallelism, and TeaCache. Available visual generation endpoints include ``/v1/images/generations``, ``/v1/videos``, ``/v1/videos/sync`` (with ``/v1/videos/generations`` kept as a deprecated alias), and video management APIs. For full details, see the :doc:`../../models/visual-generation.md` feature documentation. Example client scripts are available in the `examples/visual_gen/serve/ `_ directory. diff --git a/docs/source/models/visual-generation.md b/docs/source/models/visual-generation.md index 21bbb35bea75..c036243e2832 100644 --- a/docs/source/models/visual-generation.md +++ b/docs/source/models/visual-generation.md @@ -107,12 +107,17 @@ When served via `trtllm-serve`, the following OpenAI-compatible endpoints are av | `/v1/images/generations` | POST | Synchronous image generation | | `/v1/images/edits` | POST | Image editing | | `/v1/videos` | POST | Asynchronous video generation | -| `/v1/videos/generations` | POST | Synchronous video generation | +| `/v1/videos/sync` | POST | Synchronous video generation | +| `/v1/videos/generations` | POST | Deprecated alias of `/v1/videos/sync` (kept for back-compat) | | `/v1/videos/{id}` | GET | Video status / metadata | | `/v1/videos/{id}/content` | GET | Download generated video | | `/v1/videos/{id}` | DELETE | Delete generated video | | `/v1/videos` | GET | List all videos | +The asynchronous `/v1/videos` job advances through `GET /v1/videos/{id}`: `queued` → `generating` (model inference) → `postprocessing` (encode the media and/or write the output file) → `completed`. The `generating` → `postprocessing` transition marks the end of inference; the video is downloadable via `/content` once `completed`. + +`response_format="path"` returns the generated file's server-side path (under `TRTLLM_MEDIA_STORAGE_PATH`) for co-located clients, enabled by default. Set `TRTLLM_DISALLOW_LOCAL_MEDIA_PATH=1` to reject such requests with HTTP 400. See the [serve examples](https://github.com/NVIDIA/TensorRT-LLM/tree/main/examples/visual_gen/serve) for the full `response_format` reference. + ## Optimizations ### Quantization diff --git a/docs/source/release-notes.md b/docs/source/release-notes.md index 7e8b3e92a29d..fbf4673ea946 100644 --- a/docs/source/release-notes.md +++ b/docs/source/release-notes.md @@ -34,6 +34,8 @@ All published functionality in the Release Notes has been fully tested and verif - **[BREAKING CHANGE] KV Cache Manager V2 reports cold-tier `secondary*` statistics in `kvCacheIterationStatsByColdPoolGroup`. These keys are no longer present in the hot pool-group or window-size views.** +- **[BREAKING CHANGE] VisualGen video `response_format`.** Video generation (`POST /v1/videos/sync`, `POST /v1/videos`) narrows `response_format` to `{file, path}` (default `file`): the old `url` value — which returned raw bytes, not a URL — is renamed to `file`, and `b64_json` is removed. Requests still sending `url` or `b64_json` now get an error that names the replacement. The internal `output_path` is no longer emitted by `GET /v1/videos/{id}` or `GET /v1/videos` (status only); a co-located client obtains the on-disk path via `response_format="path"`. Image `response_format` is unchanged and additively gains `path`. The synchronous route is now `POST /v1/videos/sync`, with `POST /v1/videos/generations` kept as a deprecated alias. `response_format="path"` returns absolute server-side file paths and can be disabled server-side with `TRTLLM_DISALLOW_LOCAL_MEDIA_PATH=1` (enabled by default; paths stay under the media-storage directory). + ### Fixed Issues ### Known Issues diff --git a/examples/visual_gen/serve/README.md b/examples/visual_gen/serve/README.md index e075f8080b2a..1a8a3019bdd7 100644 --- a/examples/visual_gen/serve/README.md +++ b/examples/visual_gen/serve/README.md @@ -146,7 +146,7 @@ python sync_video_gen.py --mode ti2v \ - `--size` - Video resolution in WxH format (default: 256x256) - `--output` - Output video file path (default: output_sync.mp4) -**API Endpoint:** `POST /v1/videos/generations` +**API Endpoint:** `POST /v1/videos/sync` **API Details:** - T2V uses JSON `Content-Type: application/json` @@ -275,7 +275,7 @@ You can customize these by: - `seed`: Random seed; `null` / omitted means the engine draws a fresh seed - `num_inference_steps`, `guidance_scale`, `max_sequence_length`, `negative_prompt`: per-request denoise controls (override pipeline defaults when sent) - `extra_params`: model-specific overflow as a JSON object (see "Model-Specific `extra_params`" below). Unknown keys are rejected by the executor. -- `response_format`: `"b64_json"` or `"url"` +- `response_format`: `"url"` (default; HTTP URL to `/content`), `"b64_json"` (inline base64), or `"path"` (server-side on-disk path, for co-located clients) - `format`: Generation content encoding. Image encoders: `"png"`, `"webp"`, `"jpeg"`. Tensor formats: `"safetensors"`, `"pt"`. - Accept-and-warn OpenAI-shape fields (no engine semantic): `model`, `quality`, `style`, `user`. Sending `quality`/`style` logs a server-side WARNING; sending `model` warns on mismatch. None of these change generation behavior. @@ -289,9 +289,11 @@ You can customize these by: - `input_reference`: Reference image (I2V/TI2V) or video (V2V), accepted as a base64-encoded string in JSON or as a file in multipart form-data - **Supported formats**: PNG and JPEG images; MP4 and AVI video, with H.264 the tested codec and others best-effort. HEIF/AVIF are not supported. - `extra_params`: model-specific overflow (see below) -- `response_format`: `"b64_json"` or `"url"` +- `response_format`: `"file"` (default; `FileResponse` byte download) or `"path"` (server-side output path JSON, for co-located clients) - `format`: Generation content encoding. Video encoders: `"mp4"`, `"avi"`, `"auto"`. Tensor formats: `"safetensors"`, `"pt"` (carries video + audio + scalar metadata in one payload for LTX-2). +> **`response_format="path"`** (image and video) returns absolute server-side file paths under the server's media-storage directory (`TRTLLM_MEDIA_STORAGE_PATH`), for clients co-located with the server (shared filesystem). Enabled by default; set `TRTLLM_DISALLOW_LOCAL_MEDIA_PATH=1` to reject `path` requests with HTTP 400. + #### Tensor-format consumer contract When `format="safetensors"` or `format="pt"`, the payload bundles every populated media tensor (`image` / `video` / `audio`) and the scalar metadata (`frame_rate`, `audio_sample_rate`) into one file. @@ -406,6 +408,8 @@ curl -X POST "http://localhost:8000/v1/videos" \ curl -X GET "http://localhost:8000/v1/videos/{video_id}" ``` +The async job's `status` advances `queued` → `generating` (model inference) → `postprocessing` (encode the media and/or write the output file) → `completed`. The `generating` → `postprocessing` transition marks the end of inference; poll for `completed` to download via `/content`. + ### Download Video ```bash # The server returns either MP4 (with ffmpeg) or AVI (without ffmpeg) @@ -426,14 +430,14 @@ curl -X DELETE "http://localhost:8000/v1/videos/{video_id}" | Endpoint | Method | Mode | Content-Type | Purpose | |----------|--------|------|--------------|---------| | `/v1/videos` | POST | Async | JSON or Multipart | Create video job (T2V/TI2V) | -| `/v1/videos/generations` | POST | Sync | JSON or Multipart | Generate video sync (T2V/TI2V) | +| `/v1/videos/sync` | POST | Sync | JSON or Multipart | Generate video sync (T2V/TI2V) | | `/v1/videos/{id}` | GET | - | - | Get video status/metadata | | `/v1/videos/{id}/content` | GET | - | - | Download video file | | `/v1/videos/{id}` | DELETE | - | - | Delete video | | `/v1/videos` | GET | - | - | List all videos | | `/v1/images/generations` | POST | - | JSON | Generate images (T2I) | -**Note:** Both `/v1/videos` (async) and `/v1/videos/generations` (sync) support: +**Note:** Both `/v1/videos` (async) and `/v1/videos/sync` (sync) support: - **JSON**: Standard text-to-video (T2V) - **Multipart/Form-Data**: Text+image-to-video (TI2V) with file upload diff --git a/examples/visual_gen/serve/sync_video_gen.py b/examples/visual_gen/serve/sync_video_gen.py index 4de8ee5072f8..385369699f3c 100755 --- a/examples/visual_gen/serve/sync_video_gen.py +++ b/examples/visual_gen/serve/sync_video_gen.py @@ -1,7 +1,7 @@ #!/usr/bin/env python """Test script for synchronous video generation endpoint. -Tests POST /v1/videos/generations endpoint which waits for completion and returns video data. +Tests POST /v1/videos/sync endpoint which waits for completion and returns video data. The video is generated synchronously and the response contains the video file. Supports two modes: @@ -65,7 +65,7 @@ def test_sync_video_generation( print(f" Size: {size}") try: - endpoint = f"{base_url}/videos/generations" + endpoint = f"{base_url}/videos/sync" if input_reference: # TI2V mode - Use multipart/form-data with file upload diff --git a/tensorrt_llm/_torch/visual_gen/models/ltx2/pipeline_ltx2_two_stages.py b/tensorrt_llm/_torch/visual_gen/models/ltx2/pipeline_ltx2_two_stages.py index 181ff81af71f..afb2bdd687e9 100644 --- a/tensorrt_llm/_torch/visual_gen/models/ltx2/pipeline_ltx2_two_stages.py +++ b/tensorrt_llm/_torch/visual_gen/models/ltx2/pipeline_ltx2_two_stages.py @@ -56,10 +56,11 @@ class _TwoStagePhaseTimer(CudaPhaseTimer): """CudaPhaseTimer + the two-stage extras: the stage-2 refinement loop and the decode section. - Inherited marks keep their contract (``denoise`` = the whole stage-1 - forward; stage 2 folds into ``post_denoise`` on ``PipelineOutput``). + ``mark_post_start`` is placed at the decode boundary, so ``denoise`` + spans stage 1 plus the stage-2 upsample / LoRA bind / refinement loop + and ``post_denoise`` is the decode only (matching single-stage models). The extra event pair brackets the stage-2 refinement step loop only - (upsample / LoRA bind / text-cache prep stay outside). + (upsample / LoRA bind / text-cache prep stay outside), for logging. Event deltas are GPU-stream distances: they include GPU work plus any CPU time exposed to the stream, and stay correct under CUDA graphs and @@ -1356,8 +1357,6 @@ def forward( audio_latents = out.audio # (B, C, F_aud, M) or None assert video_latents is not None, "stage-1 latents missing on this rank" - timer.mark_post_start() - # ================================================================ # Stage 2: spatial upsample + refinement denoise — all ranks, collectively # ================================================================ @@ -1466,6 +1465,9 @@ def forward( else: self._lora_cuda_graph_state = "original" + # Denoise (stage 1 + stage 2) complete; decode is the post phase. + timer.mark_post_start() + # ================================================================ # Decode # ================================================================ diff --git a/tensorrt_llm/serve/openai_protocol.py b/tensorrt_llm/serve/openai_protocol.py index 824634970d96..13f8c4bec5b3 100644 --- a/tensorrt_llm/serve/openai_protocol.py +++ b/tensorrt_llm/serve/openai_protocol.py @@ -1628,7 +1628,7 @@ class ImageGenerationRequest(OpenAIBaseModel): # Prompt + transport (OpenAI-standard, always honored) prompt: str - response_format: Literal["url", "b64_json"] = "url" + response_format: Literal["url", "b64_json", "path"] = "url" format: Literal["png", "webp", "jpeg", "safetensors", "pt"] = Field( default="png", description=( @@ -1762,6 +1762,7 @@ class ImageObject(OpenAIBaseModel): """Generated image object in the response.""" b64_json: Optional[str] = None url: Optional[str] = None + path: Optional[str] = None revised_prompt: Optional[str] = None @@ -1792,7 +1793,7 @@ class VideoGenerationRequest(OpenAIBaseModel): # Prompt + transport prompt: str - response_format: Literal["url", "b64_json"] = "url" + response_format: Literal["file", "path"] = "file" format: Literal["mp4", "avi", "auto", "safetensors", "pt"] = Field( default="auto", description=( @@ -1865,6 +1866,26 @@ def _check_paired_dimensions(self): f"{self.width!r}, height={self.height!r}") return self + @field_validator("response_format", mode="before") + @classmethod + def _reject_removed_response_format(cls, value): + """Give migrating callers an actionable error for removed values. + + ``url``/``b64_json`` were valid before the transport rewrite; run + before the ``Literal`` check so the error names the replacement + instead of the generic "Input should be 'file' or 'path'". + """ + removed = { + "url": + ("'url' was removed for video; use 'file' (raw bytes -- the " + "old 'url' behavior, renamed) or 'path' (server-side path)."), + "b64_json": ("'b64_json' was removed for video; use 'file' (raw " + "bytes) or 'path' (server-side path)."), + } + if isinstance(value, str) and value in removed: + raise ValueError(removed[value]) + return value + class VideoJob(OpenAIBaseModel): """Metadata for an asynchronous video generation job. @@ -1886,8 +1907,13 @@ class VideoJob(OpenAIBaseModel): default=None, description="Progress of the video generation job (0-100)") prompt: str = Field(description="The prompt used to generate the video") - status: Literal["queued", "in_progress", "completed", "failed"] = Field( - description="Current status of the video generation job") + status: Literal["queued", "generating", "postprocessing", "completed", + "failed"] = Field(description=( + "Current status of the video generation job. " + "``generating`` (model inference) becomes " + "``postprocessing`` (encode and/or write the output " + "file) when inference finishes, then ``completed`` " + "once downloadable via ``/content``.")) # Video properties duration: Optional[float] = Field(default=None, @@ -1900,17 +1926,35 @@ class VideoJob(OpenAIBaseModel): ) size: Optional[str] = Field(default=None, description="Video dimensions in 'WxH' format") + # exclude=True: internal file-location for /content resolution + delete; + # never on the wire (the path payload is the hand-built {id, output_path} + # envelope in /content), so status/list model_dump() stays status-only. output_path: Optional[str] = Field( - default=None, description="Actual path where the video file was saved") + default=None, + exclude=True, + description="Server-side saved path (internal; excluded from the wire)." + ) output_paths: Optional[List[str]] = Field( - default=None, description="Paths for all generated videos when n > 1") - response_format: Optional[Literal["url", "b64_json"]] = Field( + default=None, + exclude=True, + description= + "Server-side paths for n>1 (internal; excluded from the wire).") + # exclude=True internal timings, never on the wire (status/list + # model_dump() stays status-only). ``request_started`` is a + # ``perf_counter()`` stamped at the POST handler; the background task + # computes ``total`` from it and stores the header timings + # (``generation``/``denoise``/``total``) in ``timing_metrics`` so + # ``/content`` emits the same Server-Timing header as the sync route. + request_started: Optional[float] = Field(default=None, exclude=True) + timing_metrics: Optional[Dict[str, float]] = Field(default=None, + exclude=True) + response_format: Optional[Literal["file", "path"]] = Field( default=None, description=( "Transport the client requested. ``GET /v1/videos/{id}/content`` " - "honors this: ``b64_json`` returns the encoded payload as a " - "base64 string inside a JSON envelope; ``url`` (or unset) " - "returns the file as a ``FileResponse`` download."), + "honors this: ``path`` returns the server-side output path(s) in a " + "JSON envelope; ``file`` (or unset) returns the file as a " + "``FileResponse`` download."), ) diff --git a/tensorrt_llm/serve/openai_server.py b/tensorrt_llm/serve/openai_server.py index 763ec4d8360c..366734f6c06a 100644 --- a/tensorrt_llm/serve/openai_server.py +++ b/tensorrt_llm/serve/openai_server.py @@ -102,8 +102,8 @@ from tensorrt_llm.serve.responses_utils import \ request_preprocess as responses_api_request_preprocess from tensorrt_llm.serve.tool_parser.tool_parser_factory import ToolParserFactory -from tensorrt_llm.serve.visual_gen_metrics import \ - build_visual_gen_timing_headers +from tensorrt_llm.serve.visual_gen_metrics import ( + build_visual_gen_server_timings, build_visual_gen_timing_headers) from tensorrt_llm.serve.visual_gen_utils import ( cleanup_materialized_conditioning_inputs, parse_visual_gen_params) from tensorrt_llm.version import __version__ as VERSION @@ -1258,9 +1258,15 @@ def register_visual_gen_routes(self): self.openai_video_generation_async, methods=["POST"]) # Synchronous video generation (waits for completion, extended API) - self.app.add_api_route("/v1/videos/generations", + self.app.add_api_route("/v1/videos/sync", self.openai_video_generation_sync, methods=["POST"]) + # Deprecated alias of /v1/videos/sync, retained for upstream + # back-compat after the rename; both hit the same sync handler. + self.app.add_api_route("/v1/videos/generations", + self.openai_video_generation_sync, + methods=["POST"], + deprecated=True) # Video management endpoints self.app.add_api_route("/v1/videos", self.list_videos, methods=["GET"]) self.app.add_api_route("/v1/videos/{video_id}", @@ -2521,6 +2527,10 @@ async def openai_image_generation(self, request: ImageGenerationRequest, try: image_id = f"image_{uuid.uuid4().hex}" + path_error = self._reject_disabled_path(request.response_format) + if path_error is not None: + return path_error + # Client-side ValueErrors from request translation and # parameter validation are 400. Serialization failures below # (server-side: missing media, inconsistent batch) fall @@ -2572,13 +2582,13 @@ async def openai_image_generation(self, request: ImageGenerationRequest, self.media_storage_path / f"{image_id}_{i}{ext}" for i in range(batch_size) ] - output.save(paths_in, format=request.format) + # Report the paths save() actually wrote, not paths_in -- + # save() may normalize (e.g. fill a missing extension), so + # its return is the on-disk location the client will open(). + saved = output.save(paths_in, format=request.format) data = [ - ImageObject( - url=self._build_image_content_url( - raw_request, image_id, i), - revised_prompt=request.prompt, - ) for i in range(batch_size) + self._image_object(request, raw_request, image_id, i, + saved[i]) for i in range(batch_size) ] response = ImageGenerationResponse( created=int(time.time()), @@ -2610,11 +2620,8 @@ async def openai_image_generation(self, request: ImageGenerationRequest, path.write_bytes( image_to_bytes(image, format=pil_format)) data.append( - ImageObject( - url=self._build_image_content_url( - raw_request, image_id, i), - revised_prompt=request.prompt, - )) + self._image_object(request, raw_request, image_id, + i, path)) response = ImageGenerationResponse( created=int(time.time()), data=data, @@ -2629,7 +2636,8 @@ async def openai_image_generation(self, request: ImageGenerationRequest, logger.info(f"Image {image_id} generated and encoded: " f"latency={latency:.3f}s generation={generation:.3f}s " f"denoise={denoise:.3f}s") - headers = build_visual_gen_timing_headers(metrics) + headers = build_visual_gen_timing_headers( + build_visual_gen_server_timings(metrics)) return JSONResponse(content=response.model_dump(), headers=headers) @@ -2679,6 +2687,48 @@ def _build_image_content_url(raw_request: Request, image_id: str, base = str(raw_request.base_url).rstrip("/") return f"{base}/v1/images/{image_id}/content?i={i}" + def _reject_disabled_path( + self, response_format: Optional[str]) -> Optional[Response]: + """Return a 400 when ``response_format='path'`` but it is disabled. + + ``path`` discloses absolute server-side filesystem paths, so it can be + turned off via ``TRTLLM_DISALLOW_LOCAL_MEDIA_PATH=1`` on shared / + untrusted deployments (enabled by default). Returns ``None`` when + allowed. + """ + if response_format != "path": + return None + raw = os.environ.get("TRTLLM_DISALLOW_LOCAL_MEDIA_PATH", "0") + if raw not in ("0", "1"): + logger.warning( + "Unrecognized value for TRTLLM_DISALLOW_LOCAL_MEDIA_PATH: " + f"{raw!r}. Expected '0' or '1'. Treating as '0' " + "(response_format='path' enabled).") + if raw == "1": + return self.create_error_response( + "response_format='path' is disabled on this server " + "(TRTLLM_DISALLOW_LOCAL_MEDIA_PATH=1); it returns " + "server-side filesystem paths and is only meaningful for " + "co-located clients.", + err_type="BadRequestError", + status_code=HTTPStatus.BAD_REQUEST, + ) + return None + + def _image_object(self, request: ImageGenerationRequest, + raw_request: Request, image_id: str, i: int, + path: Path) -> ImageObject: + """Build the per-item ``ImageObject`` for the ``path``/``url`` transports. + + ``b64_json`` is handled separately. Shared by the tensor and encoder + branches so they cannot drift when a transport changes. + """ + if request.response_format == "path": + return ImageObject(path=str(path), revised_prompt=request.prompt) + return ImageObject(url=self._build_image_content_url( + raw_request, image_id, i), + revised_prompt=request.prompt) + async def _parse_image_edit_request( self, raw_request: Request, @@ -2820,7 +2870,8 @@ async def openai_image_edit(self, raw_request: Request) -> Response: logger.info(f"Image {image_id} edited and encoded: " f"latency={latency:.3f}s generation={generation:.3f}s " f"denoise={denoise:.3f}s") - headers = build_visual_gen_timing_headers(metrics) + headers = build_visual_gen_timing_headers( + build_visual_gen_server_timings(metrics)) return JSONResponse(content=response.model_dump(), headers=headers) diff --git a/tensorrt_llm/serve/openai_video_routes.py b/tensorrt_llm/serve/openai_video_routes.py index a409e8f701ff..d8dea240196d 100644 --- a/tensorrt_llm/serve/openai_video_routes.py +++ b/tensorrt_llm/serve/openai_video_routes.py @@ -13,7 +13,6 @@ from __future__ import annotations import asyncio -import base64 import json import os import time @@ -21,7 +20,7 @@ import uuid from http import HTTPStatus from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Optional, Union from fastapi import Request from fastapi.responses import FileResponse, JSONResponse, Response @@ -31,7 +30,10 @@ from tensorrt_llm.media.encoding import resolve_video_format from tensorrt_llm.media.tensor_payload import is_tensor_format from tensorrt_llm.serve.openai_protocol import VideoGenerationRequest, VideoJob, VideoJobList -from tensorrt_llm.serve.visual_gen_metrics import build_visual_gen_timing_headers +from tensorrt_llm.serve.visual_gen_metrics import ( + build_visual_gen_server_timings, + build_visual_gen_timing_headers, +) from tensorrt_llm.serve.visual_gen_utils import VIDEO_STORE, parse_visual_gen_params if TYPE_CHECKING: @@ -72,18 +74,15 @@ def _preflight_encoder_format(fmt): raise ValueError(str(exc)) from exc -def _b64_json_video_response(video_id: str, fmt: str, path: Path) -> JSONResponse: - """Build the OpenAI-style ``{id, format, b64_json}`` envelope. +def _path_json_video_response( + video_id: str, path: Union[str, Path], headers: Optional[dict[str, str]] = None +) -> JSONResponse: + """Build the ``{id, output_path}`` path-transport envelope. - Reads bytes from a saved video file on disk and base64-inlines them. + Returns the server-side output path so a co-located client reads the file + directly. ``headers`` (Server-Timing metrics) are attached to the response. """ - return JSONResponse( - content={ - "id": video_id, - "format": fmt, - "b64_json": base64.b64encode(path.read_bytes()).decode("utf-8"), - } - ) + return JSONResponse(content={"id": video_id, "output_path": str(path)}, headers=headers) class _VideoRoutesMixin: @@ -104,6 +103,9 @@ async def openai_video_generation_sync(self, raw_request: Request) -> Response: - JSON: Send VideoGenerationRequest as application/json - Multipart: Send form fields + optional input_reference file """ + # Stamp request arrival for the Server-Timing ``total`` (full server + # time, incl. request parsing) before any work. + request_received = time.perf_counter() try: # Client-side ValueErrors from content-type parsing, request # translation, encoder-format preflight, parameter validation, @@ -113,6 +115,9 @@ async def openai_video_generation_sync(self, raw_request: Request) -> Response: try: # Parse request based on content-type request = await self._parse_video_generation_request(raw_request) + path_error = self._reject_disabled_path(request.response_format) + if path_error is not None: + return path_error video_id = f"video_{uuid.uuid4().hex}" params = parse_visual_gen_params( request, @@ -166,9 +171,15 @@ async def openai_video_generation_sync(self, raw_request: Request) -> Response: f"Video {video_id} serialized as tensor: latency={latency:.3f}s " f"generation={getattr(output.metrics, 'generation', 0.0):.3f}s" ) - if request.response_format == "b64_json": - return _b64_json_video_response(video_id, request.format, target) - return FileResponse(str(target), media_type=media_type, filename=target.name) + total = time.perf_counter() - request_received + headers = build_visual_gen_timing_headers( + build_visual_gen_server_timings(output.metrics, total=total) + ) + if request.response_format == "path": + return _path_json_video_response(video_id, target, headers) + return FileResponse( + str(target), media_type=media_type, filename=target.name, headers=headers + ) # Encoder formats: one file per item; ship the first item as # the route's primary download (OpenAI sync video API does @@ -199,16 +210,17 @@ async def openai_video_generation_sync(self, raw_request: Request) -> Response: f"latency={latency:.3f}s generation={generation:.3f}s " f"denoise={denoise:.3f}s" ) - headers = build_visual_gen_timing_headers(metrics) + total = time.perf_counter() - request_received + headers = build_visual_gen_timing_headers( + build_visual_gen_server_timings(metrics, total=total) + ) # TODO(TRTLLM-11579): the OpenAI Videos API does not yet define a # multi-file response, so we return only the first video as a file # download while persisting all of them to disk. actual_path = saved_paths[0] - if request.response_format == "b64_json": - return _b64_json_video_response( - video_id, actual_path.suffix.lstrip("."), actual_path - ) + if request.response_format == "path": + return _path_json_video_response(video_id, actual_path, headers) return FileResponse( str(actual_path), media_type=_video_content_type(actual_path.suffix), @@ -322,9 +334,16 @@ async def openai_video_generation_async( - JSON: Send VideoGenerationRequest as application/json - Multipart: Send form fields + optional input_reference file """ + # Stamp request arrival for the Server-Timing ``total`` (full server + # time, POST arrival -> job completed); the background task reads it + # back off the job to compute ``total``. + request_received = time.perf_counter() try: # Parse request based on content-type request = await self._parse_video_generation_request(raw_request) + path_error = self._reject_disabled_path(request.response_format) + if path_error is not None: + return path_error video_id = f"video_{uuid.uuid4().hex}" params = parse_visual_gen_params( @@ -359,6 +378,7 @@ async def openai_video_generation_async( fps=params.frame_rate, size=f"{params.width}x{params.height}", response_format=request.response_format, + request_started=request_received, ) await VIDEO_STORE.upsert(video_id, video_job) @@ -397,6 +417,10 @@ async def _generate_video_background( """Background task to generate video and save to storage.""" try: background_start = time.perf_counter() + job = await VIDEO_STORE.get(video_id) + if job: + job.status = "generating" + await VIDEO_STORE.upsert(video_id, job) future = self.generator.generate_async(inputs=request.prompt, params=params) output = await future @@ -410,6 +434,13 @@ async def _generate_video_background( await VIDEO_STORE.upsert(video_id, job) return + # Generation finished, postprocessing starts: expose the transition + # so clients can measure pure generation time on their side. + job = await VIDEO_STORE.get(video_id) + if job: + job.status = "postprocessing" + await VIDEO_STORE.upsert(video_id, job) + if is_tensor_format(request.format): # One tensor file per batch item, mirroring the encoder # path; the async job records all paths on @@ -424,11 +455,19 @@ async def _generate_video_background( resolved_fmt, _ = resolve_video_format(request.format) batch_size = output.video.shape[0] if output.video.dim() == 5 else 1 paths_in = [self.media_storage_path / f"{video_id}_{i}" for i in range(batch_size)] - saved_paths = output.save( - paths_in, + _save_kwargs = dict( format=resolved_fmt, frame_rate=output.frame_rate or request.frame_rate or params.frame_rate, ) + if os.environ.get("TRTLLM_VIDEO_ASYNC_ENCODE", "1") != "0": + # Offload the blocking encode to a thread so the event loop + # stays responsive during ``postprocessing`` — pollers can + # observe the state and other requests progress meanwhile. + saved_paths = await asyncio.get_running_loop().run_in_executor( + None, lambda: output.save(paths_in, **_save_kwargs) + ) + else: + saved_paths = output.save(paths_in, **_save_kwargs) latency = time.perf_counter() - background_start # seconds metrics = output.metrics generation = metrics.generation if metrics is not None else 0.0 @@ -442,12 +481,19 @@ async def _generate_video_background( if job: job.status = "completed" job.completed_at = int(time.time()) - # TODO: Expose VisualGen timing metrics for async jobs once the - # OpenAI video job metadata contract includes server timings. # Store the first path on output_path for single-video # compatibility, and the full list on output_paths. job.output_path = str(saved_paths[0]) job.output_paths = [str(p) for p in saved_paths] + # Timings dict for the /content Server-Timing header (excluded + # from the status wire). ``total`` spans POST arrival -> + # completion. + total = ( + time.perf_counter() - job.request_started + if job.request_started is not None + else None + ) + job.timing_metrics = build_visual_gen_server_timings(metrics, total=total) await VIDEO_STORE.upsert(video_id, job) except Exception as e: @@ -473,6 +519,8 @@ async def list_videos(self, raw_request: Request) -> Response: response = VideoJobList( data=video_jobs, ) + # output_path/output_paths are Field(exclude=True) on VideoJob, so + # model_dump() drops them from every listed job automatically. return JSONResponse(content=response.model_dump()) except Exception as e: @@ -508,6 +556,9 @@ async def get_video_metadata(self, video_id: str, raw_request: Request) -> Respo status_code=HTTPStatus.BAD_REQUEST, ) + # Status-only: output_path/output_paths are Field(exclude=True) on + # VideoJob, so model_dump() drops them automatically (no path leak); + # the fields stay on the job for /content resolution + delete. return JSONResponse(content=job.model_dump()) except Exception as e: @@ -568,13 +619,15 @@ async def get_video_content(self, video_id: str, raw_request: Request) -> Respon if video_path and os.path.exists(video_path): suffix = video_path.suffix.lstrip(".") + # Same Server-Timing header as the sync route, rebuilt from the + # timings the background task stored on the job. + headers = build_visual_gen_timing_headers(job.timing_metrics) # When the original ``POST /v1/videos`` requested - # ``response_format="b64_json"``, return the bytes - # as a base64 envelope so the async transport - # matches what the sync route does for the same - # ``response_format``. - if job.response_format == "b64_json": - return _b64_json_video_response(video_id, suffix, video_path) + # ``response_format="path"``, return the server-side output + # path as JSON instead of the bytes; the sync route honors + # the same ``response_format`` value. + if job.response_format == "path": + return _path_json_video_response(video_id, video_path, headers) if is_tensor_format(suffix): media_type = "application/octet-stream" else: @@ -583,6 +636,7 @@ async def get_video_content(self, video_id: str, raw_request: Request) -> Respon video_path, media_type=media_type, filename=video_path.name, + headers=headers, ) else: return self.create_error_response( diff --git a/tensorrt_llm/serve/scripts/benchmark_visual_gen.py b/tensorrt_llm/serve/scripts/benchmark_visual_gen.py index 10463fdbb675..2ff052cefdbd 100644 --- a/tensorrt_llm/serve/scripts/benchmark_visual_gen.py +++ b/tensorrt_llm/serve/scripts/benchmark_visual_gen.py @@ -217,7 +217,7 @@ async def async_request_video_generation( pbar: Optional[tqdm] = None, session: Optional[aiohttp.ClientSession] = None, ) -> VisualGenRequestOutput: - """POST /v1/videos/generations (sync endpoint) and measure E2E latency.""" + """POST /v1/videos/sync (sync endpoint) and measure E2E latency.""" payload = _build_payload_common(request_input) payload["seconds"] = request_input.seconds payload["fps"] = request_input.fps @@ -391,7 +391,7 @@ def main(args: argparse.Namespace): endpoint_map = { "openai-images": "/v1/images/generations", - "openai-videos": "/v1/videos/generations", + "openai-videos": "/v1/videos/sync", } endpoint = args.endpoint or endpoint_map.get(backend) if endpoint is None: diff --git a/tensorrt_llm/serve/visual_gen_metrics.py b/tensorrt_llm/serve/visual_gen_metrics.py index 3e05dbb3e323..418a2abb9530 100644 --- a/tensorrt_llm/serve/visual_gen_metrics.py +++ b/tensorrt_llm/serve/visual_gen_metrics.py @@ -1,10 +1,10 @@ # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -"""Shared metadata names for VisualGen serving metrics.""" +"""Serving-layer VisualGen metrics: metric names, timings flattener, and header formatter.""" from __future__ import annotations -from typing import TYPE_CHECKING, Optional +from typing import TYPE_CHECKING, Dict, Optional if TYPE_CHECKING: from tensorrt_llm.visual_gen.output import VisualGenMetrics @@ -12,24 +12,36 @@ SERVER_TIMING_HEADER = "Server-Timing" VISUAL_GEN_DENOISE_TIMING = "denoise" VISUAL_GEN_GENERATION_TIMING = "generation" +VISUAL_GEN_TOTAL_TIMING = "total" + + +def build_visual_gen_server_timings( + metrics: Optional["VisualGenMetrics"] = None, + total: Optional[float] = None, +) -> Dict[str, float]: + """Flatten engine ``generation``/``denoise`` + serve ``total`` into one timings dict (seconds). + + Engine metrics carry no ``total`` — the route measures it — so it is passed + separately here. Absent values are omitted. Mirrors how the LLM serve path + merges engine + server timings into one record. + """ + timings: Dict[str, float] = {} + if metrics is not None: + timings[VISUAL_GEN_GENERATION_TIMING] = metrics.generation + timings[VISUAL_GEN_DENOISE_TIMING] = metrics.denoise + if total is not None: + timings[VISUAL_GEN_TOTAL_TIMING] = total + return timings def _server_timing_metric(name: str, duration_seconds: float) -> str: - # Server-Timing ``dur`` is in milliseconds; VisualGenMetrics stores seconds. + # Server-Timing ``dur`` is in milliseconds; timings are stored in seconds. return f"{name};dur={duration_seconds * 1000:.6f}" -def build_visual_gen_timing_headers( - metrics: Optional["VisualGenMetrics"], -) -> dict[str, str]: - """Build standard Server-Timing headers for VisualGen engine timings.""" - if metrics is None: +def build_visual_gen_timing_headers(timings: Optional[Dict[str, float]]) -> dict[str, str]: + """Format a timings dict as a ``Server-Timing`` header (``{}`` if empty).""" + if not timings: return {} - return { - SERVER_TIMING_HEADER: ", ".join( - [ - _server_timing_metric(VISUAL_GEN_GENERATION_TIMING, metrics.generation), - _server_timing_metric(VISUAL_GEN_DENOISE_TIMING, metrics.denoise), - ] - ) - } + parts = [_server_timing_metric(name, dur) for name, dur in timings.items() if dur is not None] + return {SERVER_TIMING_HEADER: ", ".join(parts)} if parts else {} diff --git a/tests/unittest/_torch/visual_gen/test_trtllm_serve_e2e.py b/tests/unittest/_torch/visual_gen/test_trtllm_serve_e2e.py index 4c59c30decbe..4be47671a28a 100644 --- a/tests/unittest/_torch/visual_gen/test_trtllm_serve_e2e.py +++ b/tests/unittest/_torch/visual_gen/test_trtllm_serve_e2e.py @@ -247,9 +247,9 @@ def test_health(self, server): ], ) def test_t2v_sync(self, server, format_, expected_content_type): - """Synchronous text-to-video via POST /v1/videos/generations.""" + """Synchronous text-to-video via POST /v1/videos/sync.""" resp = requests.post( - server.url_for("v1", "videos", "generations"), + server.url_for("v1", "videos", "sync"), json={ "prompt": "A cute cat playing piano", "size": "480x320", @@ -367,10 +367,10 @@ def test_health(self, server): ], ) def test_ti2v_sync(self, server, format_, expected_content_type): - """Synchronous image-to-video via multipart POST /v1/videos/generations.""" + """Synchronous image-to-video via multipart POST /v1/videos/sync.""" with open(_REF_IMAGE_PATH, "rb") as f: resp = requests.post( - server.url_for("v1", "videos", "generations"), + server.url_for("v1", "videos", "sync"), data={ "prompt": "The cat starts playing piano, keys moving", "size": "480x320", diff --git a/tests/unittest/_torch/visual_gen/test_trtllm_serve_endpoints.py b/tests/unittest/_torch/visual_gen/test_trtllm_serve_endpoints.py index cf87cefa25c8..f238019b2333 100644 --- a/tests/unittest/_torch/visual_gen/test_trtllm_serve_endpoints.py +++ b/tests/unittest/_torch/visual_gen/test_trtllm_serve_endpoints.py @@ -5,7 +5,7 @@ POST /v1/images/generations POST /v1/images/edits - POST /v1/videos/generations (sync) + POST /v1/videos/sync (sync) POST /v1/videos (async) GET /v1/videos (list) GET /v1/videos/{video_id} (metadata) @@ -17,12 +17,15 @@ import base64 import json import os +import time from io import BytesIO from pathlib import Path from typing import Optional from unittest.mock import patch +import httpx import pytest +import pytest_asyncio import torch from fastapi.testclient import TestClient from PIL import Image @@ -116,6 +119,47 @@ def _assert_visual_gen_server_timing(headers) -> None: assert "denoise;dur=750.000000" in server_timing +def _server_timing_ms(headers, name: str) -> float: + """Parse the ``dur`` (ms) of one metric out of the Server-Timing header.""" + server_timing = headers[SERVER_TIMING_HEADER] + for part in server_timing.split(","): + part = part.strip() + if part.startswith(f"{name};dur="): + return float(part[len(f"{name};dur=") :]) + raise AssertionError(f"{name!r} not in Server-Timing: {server_timing!r}") + + +def _drive_job_to_completion(client, video_id, timeout: float = 5.0): + """Poll ``GET /v1/videos/{id}`` until the job reaches a terminal state. + + Returns the terminal status (``"completed"``/``"failed"``) or ``None`` on + timeout. Shared by the async-video tests so the polling deadline lives in + one place. + """ + deadline = time.time() + timeout + while time.time() < deadline: + status = client.get(f"/v1/videos/{video_id}").json().get("status") + if status in ("completed", "failed"): + return status + time.sleep(0.05) + return None + + +async def _adrive_job_to_completion(client, video_id, timeout: float = 10.0): + """Async counterpart of :func:`_drive_job_to_completion` for the httpx + ``AsyncClient`` — awaits polls on the live loop so the background task's + offloaded encode can progress to a terminal state. + """ + deadline = time.time() + timeout + while time.time() < deadline: + resp = await client.get(f"/v1/videos/{video_id}") + status = resp.json().get("status") + if status in ("completed", "failed"): + return status + await asyncio.sleep(0.05) + return None + + # --------------------------------------------------------------------------- # Mock VisualGen # --------------------------------------------------------------------------- @@ -316,7 +360,10 @@ def result(self, timeout=None): # --------------------------------------------------------------------------- -def _create_server(generator: MockVisualGen, model_name: str = "test-model") -> TestClient: +def _create_server( + generator: MockVisualGen, + model_name: str = "test-model", +) -> TestClient: """Instantiate an OpenAIServer for VISUAL_GEN with a mocked generator. The server detects VisualGen generators via ``_is_visual_gen_instance`` @@ -344,6 +391,34 @@ def _create_server(generator: MockVisualGen, model_name: str = "test-model") -> return client +def _create_async_client(generator: MockVisualGen, model_name: str = "test-model"): + """Build the VISUAL_GEN server and return an ``httpx.AsyncClient`` over its + ASGI app. + + The sync ``TestClient`` drives each request through a portal and does not run + the event loop between requests, so a detached ``/v1/videos`` background task + (and its offloaded encode) never progresses. Exercising the app on the + caller's live loop lets the background task run to completion. + """ + from tensorrt_llm.llmapi.disagg_utils import ServerRole + from tensorrt_llm.serve.openai_server import OpenAIServer + + with patch( + "tensorrt_llm.serve.openai_server._is_visual_gen_instance", + return_value=True, + ): + server = OpenAIServer( + generator=generator, + model=model_name, + tool_parser=None, + server_role=ServerRole.VISUAL_GEN, + metadata_server_cfg=None, + ) + return httpx.AsyncClient( + transport=httpx.ASGITransport(app=server.app), base_url="http://testserver" + ) + + # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @@ -369,6 +444,21 @@ def video_client(tmp_path): os.environ.pop("TRTLLM_MEDIA_STORAGE_PATH", None) +@pytest_asyncio.fixture() +async def async_video_client(tmp_path): + """Async httpx client over the video server — drives the async + ``/v1/videos`` background task (incl. its offloaded encode) on a live loop. + """ + gen = MockVisualGen(video_output=_make_dummy_video_tensor()) + os.environ["TRTLLM_MEDIA_STORAGE_PATH"] = str(tmp_path) + client = _create_async_client(gen) + try: + yield client + finally: + await client.aclose() + os.environ.pop("TRTLLM_MEDIA_STORAGE_PATH", None) + + @pytest.fixture() def video_audio_client(tmp_path): """TestClient backed by a MockVisualGen that produces videos with audio.""" @@ -422,6 +512,49 @@ def _dummy_save_encoded_video(video, audio, output_path, frame_rate, audio_sampl yield +@pytest.mark.parametrize( + "endpoint,payload", + [ + ("/v1/images/generations", {"prompt": "cat", "response_format": "path"}), + ( + "/v1/videos/sync", + { + "prompt": "cat", + "size": "64x64", + "seconds": 1.0, + "fps": 8, + "response_format": "path", + }, + ), + ( + "/v1/videos", + { + "prompt": "cat", + "size": "64x64", + "seconds": 1.0, + "fps": 8, + "response_format": "path", + }, + ), + ], +) +def test_response_format_path_rejected_when_disabled(tmp_path, monkeypatch, endpoint, payload): + """With ``TRTLLM_DISALLOW_LOCAL_MEDIA_PATH=1``, ``response_format='path'`` + is rejected with 400 on the image and video (sync + async) endpoints.""" + monkeypatch.setenv("TRTLLM_DISALLOW_LOCAL_MEDIA_PATH", "1") + monkeypatch.setenv("TRTLLM_MEDIA_STORAGE_PATH", str(tmp_path)) + gen = MockVisualGen( + image_output=_make_dummy_image_tensor(), + video_output=_make_dummy_video_tensor(), + ) + client = _create_server(gen) + resp = client.post(endpoint, json=payload, headers={"content-type": "application/json"}) + assert resp.status_code == 400 + body = resp.json() + _assert_llm_envelope(body, code=400) + assert "path" in body["message"] and "disabled" in body["message"] + + # ========================================================================= # POST /v1/images/generations # ========================================================================= @@ -545,6 +678,47 @@ def test_image_generation_pt_url(self, image_client): loaded = torch.load(BytesIO(content.content), weights_only=True) assert "image" in loaded + def test_image_generation_path_returns_output_path(self, image_client): + """``response_format='path'`` writes each image to media storage and + surfaces its on-disk path per ``data[]`` item; ``n>1`` fans out to one + object per image (distinct path); ``url``/``b64_json`` stay unset.""" + resp = image_client.post( + "/v1/images/generations", + json={ + "prompt": "A dog", + "response_format": "path", + "n": 2, + }, + ) + assert resp.status_code == 200 + data = resp.json()["data"] + assert len(data) == 2 + assert len({obj["path"] for obj in data}) == 2 # one distinct path per image + for obj in data: + assert obj["url"] is None and obj["b64_json"] is None + assert obj["path"] is not None and os.path.exists(obj["path"]) + with open(data[0]["path"], "rb") as fh: + assert fh.read().startswith(b"\x89PNG\r\n\x1a\n") + + def test_image_generation_pt_path(self, image_client): + """Tensor formats under ``response_format='path'`` persist each + per-item payload and return its on-disk path.""" + resp = image_client.post( + "/v1/images/generations", + json={ + "prompt": "Tensor dog", + "response_format": "path", + "format": "pt", + }, + ) + assert resp.status_code == 200 + obj = resp.json()["data"][0] + assert obj["path"] is not None and obj["url"] is None and obj["b64_json"] is None + assert os.path.exists(obj["path"]) + with open(obj["path"], "rb") as fh: + loaded = torch.load(BytesIO(fh.read()), weights_only=True) + assert "image" in loaded + def test_image_generation_auto_size(self, image_client): resp = image_client.post( "/v1/images/generations", @@ -1130,7 +1304,7 @@ def test_4d_batch_tensor_expanded(self): # ========================================================================= -# POST /v1/videos/generations (synchronous) +# POST /v1/videos/sync (synchronous) # ========================================================================= @@ -1138,7 +1312,7 @@ def test_4d_batch_tensor_expanded(self): class TestVideoGenerationSync: def test_basic_sync_video_generation(self, video_client): resp = video_client.post( - "/v1/videos/generations", + "/v1/videos/sync", json={ "prompt": "A rocket launching", "size": "64x64", @@ -1150,11 +1324,69 @@ def test_basic_sync_video_generation(self, video_client): assert resp.status_code == 200 assert resp.headers["content-type"] == "video/mp4" _assert_visual_gen_server_timing(resp.headers) + + def test_sync_video_server_timing_has_total(self, video_client): + """The sync Server-Timing header carries generation, denoise, and the + new ``total`` (full server time; real wall-clock, so only checked > 0).""" + resp = video_client.post( + "/v1/videos/sync", + json={ + "prompt": "timing", + "size": "32x32", + "seconds": 1.0, + "fps": 8, + "format": "avi", + }, + headers={"content-type": "application/json"}, + ) + assert resp.status_code == 200 + assert _server_timing_ms(resp.headers, "generation") == 1250.0 + assert _server_timing_ms(resp.headers, "denoise") == 750.0 + assert _server_timing_ms(resp.headers, "total") > 0 assert len(resp.content) > 0 - def test_sync_video_generation_with_params(self, video_client): + def test_deprecated_generations_alias_routes_to_sync(self, video_client): + """The pre-rename /v1/videos/generations route is kept as a deprecated + alias of /v1/videos/sync (upstream back-compat) — same handler, so it + returns the video bytes rather than 404/405.""" resp = video_client.post( "/v1/videos/generations", + json={ + "prompt": "A rocket launching", + "size": "64x64", + "seconds": 1.0, + "fps": 8, + }, + headers={"content-type": "application/json"}, + ) + assert resp.status_code == 200 + assert resp.headers["content-type"] == "video/mp4" + assert len(resp.content) > 0 + + @pytest.mark.parametrize("removed", ["url", "b64_json"]) + def test_removed_response_format_names_replacement(self, video_client, removed): + """Legacy video response_format values (url/b64_json) are rejected with + a 422 whose message names the replacement, not the generic + "Input should be 'file' or 'path'".""" + resp = video_client.post( + "/v1/videos/sync", + json={ + "prompt": "x", + "size": "64x64", + "seconds": 1.0, + "fps": 8, + "response_format": removed, + }, + headers={"content-type": "application/json"}, + ) + assert resp.status_code == 422 + body = resp.json() + _assert_llm_envelope(body, code=422, message_contains=removed) + assert "removed" in body["message"] and "file" in body["message"], body["message"] + + def test_sync_video_generation_with_params(self, video_client): + resp = video_client.post( + "/v1/videos/sync", json={ "prompt": "Ocean waves", "size": "64x64", @@ -1187,7 +1419,7 @@ def test_sync_video_generation_multipart(self, video_client, tmp_path): Image.new("RGB", (4, 4), (64, 64, 64)).save(str(ref_path)) with open(ref_path, "rb") as f: resp = video_client.post( - "/v1/videos/generations", + "/v1/videos/sync", data={ "prompt": "Mountain sunrise", "size": "64x64", @@ -1206,7 +1438,7 @@ def test_sync_video_generation_multipart_with_reference(self, video_client, tmp_ with open(ref_path, "rb") as f: resp = video_client.post( - "/v1/videos/generations", + "/v1/videos/sync", data={ "prompt": "Animate this image", "size": "64x64", @@ -1236,7 +1468,7 @@ def test_sync_video_generation_multipart_with_video_reference(self, video_client payload = _V2V_FIXTURE_MP4.read_bytes() with open(_V2V_FIXTURE_MP4, "rb") as f: resp = video_client.post( - "/v1/videos/generations", + "/v1/videos/sync", data={ "prompt": "Continue the same scene", "size": "64x64", @@ -1259,7 +1491,7 @@ def test_sync_video_generation_undecodable_reference_400(self, video_client): """Content matching no image or video container signature is rejected at the boundary.""" resp = video_client.post( - "/v1/videos/generations", + "/v1/videos/sync", data={"prompt": "x"}, files={"input_reference": ("doc.txt", BytesIO(b"not media"), "text/plain")}, ) @@ -1268,7 +1500,7 @@ def test_sync_video_generation_undecodable_reference_400(self, video_client): def test_sync_video_failure(self, failing_client): resp = failing_client.post( - "/v1/videos/generations", + "/v1/videos/sync", json={ "prompt": "Should fail", "size": "64x64", @@ -1285,7 +1517,7 @@ def test_sync_video_null_output(self, tmp_path): os.environ["TRTLLM_MEDIA_STORAGE_PATH"] = str(tmp_path) client = _create_server(gen) resp = client.post( - "/v1/videos/generations", + "/v1/videos/sync", json={"prompt": "null video", "size": "64x64", "seconds": 1.0, "fps": 8}, headers={"content-type": "application/json"}, ) @@ -1305,7 +1537,7 @@ def test_sync_video_capacity_failure_is_503(self, tmp_path, monkeypatch): monkeypatch.setenv("TRTLLM_MEDIA_STORAGE_PATH", str(tmp_path)) client = _create_server(gen) resp = client.post( - "/v1/videos/generations", + "/v1/videos/sync", json={"prompt": "big", "size": "64x64", "seconds": 1.0, "fps": 8}, headers={"content-type": "application/json"}, ) @@ -1322,7 +1554,7 @@ def test_sync_video_client_failure_is_400(self, tmp_path, monkeypatch): monkeypatch.setenv("TRTLLM_MEDIA_STORAGE_PATH", str(tmp_path)) client = _create_server(gen) resp = client.post( - "/v1/videos/generations", + "/v1/videos/sync", json={"prompt": "bad ref", "size": "64x64", "seconds": 1.0, "fps": 8}, headers={"content-type": "application/json"}, ) @@ -1331,7 +1563,7 @@ def test_sync_video_client_failure_is_400(self, tmp_path, monkeypatch): def test_sync_video_unsupported_content_type(self, video_client): resp = video_client.post( - "/v1/videos/generations", + "/v1/videos/sync", content=b"some raw bytes", headers={"content-type": "text/plain"}, ) @@ -1340,7 +1572,7 @@ def test_sync_video_unsupported_content_type(self, video_client): def test_sync_video_missing_prompt_json(self, video_client): """Missing required ``prompt`` surfaces the visual-gen 422 envelope.""" resp = video_client.post( - "/v1/videos/generations", + "/v1/videos/sync", json={"size": "64x64"}, headers={"content-type": "application/json"}, ) @@ -1352,7 +1584,7 @@ def test_sync_video_missing_prompt_multipart(self, video_client): same LLM envelope as JSON so the wire contract is identical.""" dummy_file = BytesIO(b"") resp = video_client.post( - "/v1/videos/generations", + "/v1/videos/sync", data={"size": "64x64"}, files={"_dummy": ("dummy", dummy_file, "application/octet-stream")}, ) @@ -1365,7 +1597,7 @@ def test_sync_video_multipart_rejects_unknown_field(self, video_client): the JSON path.""" dummy_file = BytesIO(b"") resp = video_client.post( - "/v1/videos/generations", + "/v1/videos/sync", data={ "prompt": "Strict multipart", "size": "64x64", @@ -1381,7 +1613,7 @@ def test_sync_video_multipart_rejects_unknown_field(self, video_client): def test_sync_video_rejects_top_level_n(self, video_client): """Sync video has no top-level ``n``; it's rejected with 422.""" resp = video_client.post( - "/v1/videos/generations", + "/v1/videos/sync", json={ "prompt": "Batch rockets", "size": "64x64", @@ -1436,6 +1668,83 @@ def test_async_video_job_metadata_fields(self, video_client): assert data["fps"] == 12 assert data["size"] == "64x64" + @pytest.mark.threadleak(enabled=False) # offloaded encode uses a worker thread + @pytest.mark.asyncio + async def test_async_video_status_transitions_generating_then_postprocessing( + self, async_video_client, monkeypatch + ): + """queued -> generating -> postprocessing -> completed, in order, so + clients can detect when generation finishes before postprocessing.""" + seen = [] + original_upsert = VIDEO_STORE.upsert + + async def _spy_upsert(video_id, job): + seen.append(job.status) + return await original_upsert(video_id, job) + + monkeypatch.setattr(VIDEO_STORE, "upsert", _spy_upsert) + + resp = await async_video_client.post( + "/v1/videos", + json={"prompt": "lifecycle", "size": "32x32", "seconds": 1.0, "fps": 8}, + headers={"content-type": "application/json"}, + ) + assert resp.status_code == 202 + video_id = resp.json()["id"] + + status = await _adrive_job_to_completion(async_video_client, video_id) + assert status == "completed" + + assert "generating" in seen and "postprocessing" in seen + assert seen.index("generating") < seen.index("postprocessing") < seen.index("completed") + + @pytest.mark.threadleak(enabled=False) # offloaded encode uses a worker thread + @pytest.mark.asyncio + async def test_async_postprocessing_state_observable_during_encode( + self, async_video_client, monkeypatch + ): + """The encode is offloaded to a thread, so the event loop stays + responsive and a poll observes ``postprocessing`` while the file is + written — not just ``generating`` then ``completed``.""" + import threading + + release = threading.Event() + original_save = VisualGenOutput.save + + def _blocking_save(self, *args, **kwargs): + # Runs in the executor thread; hold until the test sees the state. + release.wait(timeout=5) + return original_save(self, *args, **kwargs) + + monkeypatch.setattr(VisualGenOutput, "save", _blocking_save) + + resp = await async_video_client.post( + "/v1/videos", + json={ + "prompt": "observe postprocessing", + "size": "32x32", + "seconds": 1.0, + "fps": 8, + "format": "auto", + }, + headers={"content-type": "application/json"}, + ) + assert resp.status_code == 202 + video_id = resp.json()["id"] + + observed = None + deadline = time.time() + 5 + while time.time() < deadline: + poll = await async_video_client.get(f"/v1/videos/{video_id}") + observed = poll.json().get("status") + if observed in ("postprocessing", "completed", "failed"): + break + await asyncio.sleep(0.02) + release.set() # never leave the encoder blocked + assert observed == "postprocessing", ( + f"GET never observed 'postprocessing' (saw {observed!r})" + ) + def test_async_video_multipart(self, video_client, tmp_path): """Multipart async request with a real ``input_reference`` file.""" ref_path = tmp_path / "ref.png" @@ -1580,6 +1889,9 @@ def test_list_videos_after_creation(self, video_client): assert resp.status_code == 200 data = resp.json() assert len(data["data"]) == 2 + # Status-only listing: the internal path fields never appear on the wire. + for item in data["data"]: + assert "output_path" not in item and "output_paths" not in item # ========================================================================= @@ -1588,20 +1900,29 @@ def test_list_videos_after_creation(self, video_client): class TestGetVideoMetadata: - def test_get_video_metadata_success(self, video_client): - create_resp = video_client.post( + @pytest.mark.threadleak(enabled=False) # offloaded encode uses a worker thread + @pytest.mark.asyncio + async def test_get_video_metadata_success(self, async_video_client): + create_resp = await async_video_client.post( "/v1/videos", json={"prompt": "Space walk", "size": "64x64", "seconds": 1.0, "fps": 8}, headers={"content-type": "application/json"}, ) video_id = create_resp.json()["id"] - resp = video_client.get(f"/v1/videos/{video_id}") + # Drive to completion so output_path/output_paths are populated on the + # job, then confirm the status endpoint returns status only (no leak). + await _adrive_job_to_completion(async_video_client, video_id) + + resp = await async_video_client.get(f"/v1/videos/{video_id}") assert resp.status_code == 200 data = resp.json() assert data["id"] == video_id assert data["object"] == "video" assert data["prompt"] == "Space walk" + assert data["status"] == "completed" + # Status-only: the internal server path(s) are not leaked here. + assert "output_path" not in data and "output_paths" not in data def test_get_video_metadata_not_found(self, video_client): resp = video_client.get("/v1/videos/video_nonexistent") @@ -1616,10 +1937,8 @@ def test_get_video_metadata_not_found(self, video_client): @pytest.mark.threadleak(enabled=False) # FileResponse spawns AnyIO worker threads class TestGetVideoContent: def _insert_video_job(self, video_id: str, status: str = "queued"): - import time as _time - job = VideoJob( - created_at=int(_time.time()), + created_at=int(time.time()), id=video_id, model="test-model", prompt="test prompt", @@ -1662,6 +1981,20 @@ def test_get_video_content_not_ready(self, tmp_path): assert resp.status_code == 400 os.environ.pop("TRTLLM_MEDIA_STORAGE_PATH", None) + @pytest.mark.parametrize("status", ["generating", "postprocessing"]) + def test_get_video_content_not_ready_in_flight(self, tmp_path, status): + """A generating/postprocessing job is not downloadable yet → 400.""" + gen = MockVisualGen(video_output=_make_dummy_video_tensor()) + os.environ["TRTLLM_MEDIA_STORAGE_PATH"] = str(tmp_path) + client = _create_server(gen) + + video_id = f"video_{status}" + self._insert_video_job(video_id, status=status) + + resp = client.get(f"/v1/videos/{video_id}/content") + assert resp.status_code == 400 + os.environ.pop("TRTLLM_MEDIA_STORAGE_PATH", None) + def test_get_video_content_completed_but_file_missing(self, tmp_path): """Video marked completed but file deleted from disk → 404.""" gen = MockVisualGen(video_output=_make_dummy_video_tensor()) @@ -1839,7 +2172,7 @@ def test_sync_video_route_renders_validation_error_at_400(self, tmp_path): ) client = _create_server(gen) resp = client.post( - "/v1/videos/generations", + "/v1/videos/sync", json={ "prompt": "trigger validation error", "size": "64x64", @@ -2017,14 +2350,14 @@ def test_visual_gen_role_uses_llm_envelope(self): @pytest.mark.threadleak(enabled=False) # FileResponse spawns AnyIO worker threads class TestVideoTensorResponse: """The sync route emits tensor payloads as a single file under - ``response_format='url'`` and as base64-encoded bytes under - ``response_format='b64_json'``. The async route persists the + ``response_format='file'`` and as a server-side path JSON under + ``response_format='path'``. The async route persists the payload to media storage; ``GET /v1/videos/{id}/content`` serves the file with ``application/octet-stream``.""" def _post_sync(self, video_client, fmt: str, response_format: str): return video_client.post( - "/v1/videos/generations", + "/v1/videos/sync", json={ "prompt": f"tensor video {fmt}", "size": "32x32", @@ -2037,8 +2370,8 @@ def _post_sync(self, video_client, fmt: str, response_format: str): ) @pytest.mark.parametrize("fmt", ["safetensors", "pt"]) - def test_sync_tensor_url_returns_file_with_correct_suffix(self, video_audio_client, fmt): - resp = self._post_sync(video_audio_client, fmt, "url") + def test_sync_tensor_file_returns_file_with_correct_suffix(self, video_audio_client, fmt): + resp = self._post_sync(video_audio_client, fmt, "file") assert resp.status_code == 200 ext = f".{fmt}" # The content-disposition header carries the on-disk filename. @@ -2054,13 +2387,17 @@ def test_sync_tensor_url_returns_file_with_correct_suffix(self, video_audio_clie assert "video" in loaded @pytest.mark.parametrize("fmt", ["safetensors", "pt"]) - def test_sync_tensor_b64_returns_decodable_payload(self, video_audio_client, fmt): - resp = self._post_sync(video_audio_client, fmt, "b64_json") + def test_sync_tensor_path_returns_readable_output_path(self, video_audio_client, fmt): + resp = self._post_sync(video_audio_client, fmt, "path") assert resp.status_code == 200 + # path responses carry the Server-Timing metrics too. + _assert_visual_gen_server_timing(resp.headers) data = resp.json() - assert data["format"] == fmt - assert "b64_json" in data - raw = base64.b64decode(data["b64_json"]) + assert set(data) >= {"id", "output_path"} + # Co-located client reads the returned server-side path directly. + assert os.path.exists(data["output_path"]) + with open(data["output_path"], "rb") as fh: + raw = fh.read() if fmt == "safetensors": from safetensors.torch import load as load_safetensors @@ -2071,8 +2408,6 @@ def test_sync_tensor_b64_returns_decodable_payload(self, video_audio_client, fmt @pytest.mark.parametrize("fmt", ["safetensors", "pt"]) def test_async_tensor_persists_and_serves(self, video_audio_client, fmt, tmp_path): - import time as _time - client = video_audio_client resp = client.post( "/v1/videos", @@ -2089,12 +2424,7 @@ def test_async_tensor_persists_and_serves(self, video_audio_client, fmt, tmp_pat video_id = resp.json()["id"] # Drive the background task to completion via polling. - deadline = _time.time() + 5 - while _time.time() < deadline: - status = client.get(f"/v1/videos/{video_id}").json().get("status") - if status in ("completed", "failed"): - break - _time.sleep(0.05) + _drive_job_to_completion(client, video_id) content = client.get(f"/v1/videos/{video_id}/content") assert content.status_code == 200 @@ -2110,44 +2440,44 @@ def test_async_tensor_persists_and_serves(self, video_audio_client, fmt, tmp_pat @pytest.mark.threadleak(enabled=False) # FileResponse spawns AnyIO worker threads -class TestVideoEncoderB64Response: +class TestVideoEncoderResponse: """The sync video route's encoder branch (``mp4``/``avi``/``auto``) - honors ``response_format='b64_json'`` by base64-encoding the - encoded video bytes; ``response_format='url'`` keeps the + honors ``response_format='path'`` by returning the server-side output + path(s) as JSON; ``response_format='file'`` keeps the ``FileResponse`` download.""" - def test_sync_encoder_b64_json_returns_base64_payload(self, video_client): + def test_sync_encoder_path_returns_output_path(self, video_client): resp = video_client.post( - "/v1/videos/generations", + "/v1/videos/sync", json={ - "prompt": "encoded b64", + "prompt": "encoded path", "size": "32x32", "seconds": 1.0, "fps": 8, "format": "avi", - "response_format": "b64_json", + "response_format": "path", }, headers={"content-type": "application/json"}, ) assert resp.status_code == 200 + # path responses carry the Server-Timing metrics too. + _assert_visual_gen_server_timing(resp.headers) body = resp.json() - assert body["format"] in {"mp4", "avi"} - assert "b64_json" in body - raw = base64.b64decode(body["b64_json"]) - # Non-empty encoded bytes — exact format verification is the - # encoder layer's domain. - assert len(raw) > 0 - - def test_sync_encoder_url_keeps_file_response(self, video_client): + assert set(body) >= {"id", "output_path"} + # The returned server-side path points at non-empty encoded bytes. + assert os.path.exists(body["output_path"]) + assert os.path.getsize(body["output_path"]) > 0 + + def test_sync_encoder_file_keeps_file_response(self, video_client): resp = video_client.post( - "/v1/videos/generations", + "/v1/videos/sync", json={ - "prompt": "encoded url", + "prompt": "encoded file", "size": "32x32", "seconds": 1.0, "fps": 8, "format": "avi", - "response_format": "url", + "response_format": "file", }, headers={"content-type": "application/json"}, ) @@ -2174,7 +2504,7 @@ class TestVideoTimingValidation: ) def test_non_positive_timing_field_rejected(self, video_client, field, value): resp = video_client.post( - "/v1/videos/generations", + "/v1/videos/sync", json={ "prompt": "bad timing", "size": "32x32", @@ -2217,7 +2547,7 @@ class TestVideoZeroFrameDerivationRejected: def test_subsecond_seconds_below_one_frame_returns_400(self, video_client): resp = video_client.post( - "/v1/videos/generations", + "/v1/videos/sync", json={ "prompt": "way too short", "size": "32x32", @@ -2241,7 +2571,7 @@ def test_seconds_without_frame_rate_returns_400(self, video_client): pipeline's default ``num_frames``.""" video_client.mock_gen.executor.default_generation_params.pop("frame_rate", None) resp = video_client.post( - "/v1/videos/generations", + "/v1/videos/sync", json={ "prompt": "duration without fps", "size": "32x32", @@ -2260,7 +2590,7 @@ def test_explicit_num_frames_one_is_accepted(self, video_client): """The caller can bypass the derivation by passing ``num_frames`` directly; the request must succeed.""" resp = video_client.post( - "/v1/videos/generations", + "/v1/videos/sync", json={ "prompt": "explicit single frame", "size": "32x32", @@ -2337,7 +2667,7 @@ def test_frame_budget_bounds(self, video_client, field, value, boundary): payload.update({"seconds": 1.0, "fps": 8}) payload[field] = value resp = video_client.post( - "/v1/videos/generations", + "/v1/videos/sync", json=payload, headers={"content-type": "application/json"}, ) @@ -2444,7 +2774,7 @@ def test_sync_route_fails_before_generate(self, video_client, monkeypatch, raise # locks in the fail-fast contract. video_client.mock_gen.last_inputs = None resp = video_client.post( - "/v1/videos/generations", + "/v1/videos/sync", json={ "prompt": "mp4 without ffmpeg", "size": "32x32", @@ -2490,7 +2820,7 @@ def test_sync_route_tensor_format_unaffected(self, video_client, monkeypatch): monkeypatch.setattr(routes, "resolve_video_format", _raise_value_error) resp = video_client.post( - "/v1/videos/generations", + "/v1/videos/sync", json={ "prompt": "tensor unaffected", "size": "32x32", @@ -2504,75 +2834,90 @@ def test_sync_route_tensor_format_unaffected(self, video_client, monkeypatch): @pytest.mark.threadleak(enabled=False) # FileResponse spawns AnyIO worker threads -class TestAsyncVideoB64JsonTransport: +class TestAsyncVideoTransport: """``POST /v1/videos`` persists the requested ``response_format`` on the queued job. ``GET /v1/videos/{id}/content`` honors it: - ``url`` (or unset) returns a ``FileResponse`` download; - ``b64_json`` returns a JSON envelope with the encoded bytes - base64-inlined.""" - - def _drive_job_to_completion(self, client, video_id): - import time as _time - - deadline = _time.time() + 5 - while _time.time() < deadline: - status = client.get(f"/v1/videos/{video_id}").json().get("status") - if status in ("completed", "failed"): - return status - _time.sleep(0.05) - return None - - def test_async_b64_json_returned_at_get_content(self, video_client): - resp = video_client.post( + ``file`` (or unset) returns a ``FileResponse`` download; + ``path`` returns a JSON envelope with the server-side output path(s).""" + + @pytest.mark.asyncio + async def test_async_path_returned_at_get_content(self, async_video_client): + resp = await async_video_client.post( "/v1/videos", json={ - "prompt": "async base64", + "prompt": "async path", "size": "32x32", "seconds": 1.0, "fps": 8, "format": "avi", - "response_format": "b64_json", + "response_format": "path", }, headers={"content-type": "application/json"}, ) assert resp.status_code == 202 job = resp.json() - assert job["response_format"] == "b64_json" + assert job["response_format"] == "path" - status = self._drive_job_to_completion(video_client, job["id"]) + status = await _adrive_job_to_completion(async_video_client, job["id"]) assert status == "completed" - content = video_client.get(f"/v1/videos/{job['id']}/content") + content = await async_video_client.get(f"/v1/videos/{job['id']}/content") assert content.status_code == 200 body = content.json() - assert set(body) >= {"id", "format", "b64_json"} + assert set(body) >= {"id", "output_path"} assert body["id"] == job["id"] - # The encoded payload decodes to non-empty bytes. - raw = base64.b64decode(body["b64_json"]) - assert len(raw) > 0 + # The returned server-side path points at non-empty bytes. + assert os.path.exists(body["output_path"]) + assert os.path.getsize(body["output_path"]) > 0 + + @pytest.mark.asyncio + async def test_async_content_server_timing_has_total(self, async_video_client): + """`/content` carries the same Server-Timing header as the sync route, + rebuilt from the timings the background task stored on the job.""" + resp = await async_video_client.post( + "/v1/videos", + json={ + "prompt": "timing", + "size": "32x32", + "seconds": 1.0, + "fps": 8, + "format": "avi", + }, + headers={"content-type": "application/json"}, + ) + assert resp.status_code == 202 + video_id = resp.json()["id"] + assert await _adrive_job_to_completion(async_video_client, video_id) == "completed" + + content = await async_video_client.get(f"/v1/videos/{video_id}/content") + assert content.status_code == 200 + assert _server_timing_ms(content.headers, "generation") == 1250.0 + assert _server_timing_ms(content.headers, "denoise") == 750.0 + assert _server_timing_ms(content.headers, "total") > 0 - def test_async_url_still_returns_file_response(self, video_client): - """Default and explicit ``response_format='url'`` keep the + @pytest.mark.asyncio + async def test_async_file_still_returns_file_response(self, async_video_client): + """Default and explicit ``response_format='file'`` keep the existing ``FileResponse`` behavior.""" - resp = video_client.post( + resp = await async_video_client.post( "/v1/videos", json={ - "prompt": "async url", + "prompt": "async file", "size": "32x32", "seconds": 1.0, "fps": 8, "format": "avi", - "response_format": "url", + "response_format": "file", }, headers={"content-type": "application/json"}, ) assert resp.status_code == 202 job = resp.json() - assert job["response_format"] == "url" + assert job["response_format"] == "file" - self._drive_job_to_completion(video_client, job["id"]) - content = video_client.get(f"/v1/videos/{job['id']}/content") + await _adrive_job_to_completion(async_video_client, job["id"]) + content = await async_video_client.get(f"/v1/videos/{job['id']}/content") assert content.status_code == 200 - # AVI FileResponse carries ``video/x-msvideo``; the b64_json + # AVI FileResponse carries ``video/x-msvideo``; the path # branch would have set ``application/json``. assert content.headers["content-type"] == "video/x-msvideo" diff --git a/tests/unittest/_torch/visual_gen/test_visual_gen_utils.py b/tests/unittest/_torch/visual_gen/test_visual_gen_utils.py index 218d580e27ee..de9d8956df4c 100644 --- a/tests/unittest/_torch/visual_gen/test_visual_gen_utils.py +++ b/tests/unittest/_torch/visual_gen/test_visual_gen_utils.py @@ -69,7 +69,7 @@ def image_request_defaults(): @pytest.fixture def video_request_defaults(): - return VideoGenerationRequest(prompt="storm", response_format="b64_json") + return VideoGenerationRequest(prompt="storm", response_format="file") # ============================================================================= diff --git a/tests/unittest/api_stability/references/trtllm_serve_api.yaml b/tests/unittest/api_stability/references/trtllm_serve_api.yaml index 34f7e05bd495..f5846dc97b25 100644 --- a/tests/unittest/api_stability/references/trtllm_serve_api.yaml +++ b/tests/unittest/api_stability/references/trtllm_serve_api.yaml @@ -1311,7 +1311,7 @@ models: required: true response_format: kind: openai - type: Literal['url', 'b64_json'] + type: Literal['url', 'b64_json', 'path'] default: url status: stable required: false @@ -1449,8 +1449,8 @@ models: required: true response_format: kind: extension - type: Literal['url', 'b64_json'] - default: url + type: Literal['file', 'path'] + default: file status: stable required: false format: