diff --git a/README.md b/README.md index 140f1c7..71b8a98 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,7 @@ HTTP API, and MCP clients all share one backend and one extension bridge. - Text-to-image and reference-image generation - Text-to-video, image-to-video, first/last-frame, reference-to-video, and video editing - 4, 6, 8, and 10-second video generation +- 1080p and 4K video delivery through Flow's upsampler, the same pass behind the Flow UI's HD download - Reusable generated and uploaded media IDs, including after backend restarts - Exact `--output` paths with real PNG, JPEG, and WebP conversion - Signature-based MIME and extension detection @@ -125,6 +126,31 @@ flow video "transition between scenes" --start first.png --end last.png flow video "keep this character consistent" --ref character.png ``` +#### Resolution + +Google Flow generates video at 720p. 1080p and 4K are a second *upsampler* pass +over the finished clip — exactly what the Flow UI does behind its +high-resolution download. Ask for it during generation: + +```bash +flow video "a dragon flying over mountains" --resolution 1080p +flow video "a dragon flying over mountains" --resolution 4k -o /absolute/path/dragon.mp4 +``` + +The upsampled file is written to `--output`; the 720p original stays in the +output directory and in `history.json`. If the upsample pass fails, the 720p +video is still delivered and the CLI prints a note. + +Upsample a clip you already generated: + +```bash +flow upsample GENERATED_VIDEO_MEDIA_ID --resolution 1080p +flow upsample previous_take.mp4 --resolution 4k -o /absolute/path/take_4k.mp4 +``` + +`1080p` upsampling is free. `4k` costs credits and needs a higher Flow tier; +the backend refuses it with HTTP 402 when the balance cannot cover it. + `--start`, `--end`, and `--ref` accept either local image paths or exact media IDs stored in `history.json`: @@ -282,7 +308,8 @@ JSON-RPC messages are sent to `http://127.0.0.1:8001/messages`. - `list_flow_models` — available models and the active default - `get_flow_history` — generated and uploaded media history - `generate_flow_image` — text/reference image generation -- `generate_flow_video` — text, start-image, and reference video generation +- `generate_flow_video` — text, start-image, and reference video generation, with `resolution` for 720p/1080p/4K delivery +- `upsample_flow_video` — upsample an existing video to 1080p or 4K - `upload_flow_media` — upload a local path, URL, or base64 media payload - `download_media_from_url` — download media and optionally upload it to Flow - `edit_flow_video` — edit a video by media ID or local video path @@ -299,12 +326,19 @@ Default base URL: `http://127.0.0.1:8001` | `GET /v1/history` | Persistent generated/uploaded media history | | `POST /v1/images/generations` | Generate images | | `POST /v1/videos/generations` | Submit video generation | -| `GET /v1/videos/generations/{job_id}` | Poll a video job | +| `GET /v1/videos/generations/{job_id}` | Poll a video or upsample job | +| `POST /v1/videos/upsample` | Upsample an existing video to 1080p or 4K | | `POST /v1/upload` | Upload an image or video reference | | `GET /download/{filename}` | Download a managed media file | | `GET /sse` | MCP over SSE | | `POST /messages` | MCP over SSE JSON-RPC messages | +`POST /v1/videos/generations` accepts `"resolution": "720p" | "1080p" | "4k"`. +Above 720p the response leads with the upsampled media and still includes the +720p original, each entry carrying `resolution` and, for upsampled files, +`source_media_id`. `POST /v1/videos/upsample` returns the same pollable job +shape and is polled through `GET /v1/videos/generations/{job_id}`. + Send an `Idempotency-Key` header when an HTTP generation request may be retried. Video submission returns a structured result containing a `job_id`, status, creation timestamp, and media data. Poll the job endpoint until it is @@ -334,6 +368,15 @@ Environment variables take precedence over values in `.env`. | `FLOW_VIDEO_POLL_TIMEOUT` | `900` | CLI video-job polling timeout | | `MAX_CONCURRENT_REQUESTS` | `5` | Maximum concurrent Flow requests | | `REQUEST_MIN_INTERVAL` | `3` | Minimum seconds between request starts | +| `VIDEO_UPSAMPLER_1080P_MODEL` | `veo_3_1_upsampler_1080p` | Flow 1080p upsampler model key | +| `VIDEO_UPSAMPLER_4K_MODEL` | `veo_3_1_upsampler_4k` | Flow 4K upsampler model key | +| `VIDEO_UPSAMPLE_ENUM_1080P` | `VIDEO_RESOLUTION_1080P` | 1080p resolution enum sent to Flow | +| `VIDEO_UPSAMPLE_ENUM_4K` | `VIDEO_RESOLUTION_4K` | 4K resolution enum sent to Flow | + +The four upsampler variables exist because Flow's upsample API is undocumented. +If Google renames a model key or a resolution enum, override it here instead of +patching code. A rejected enum is retried with the known spelling variants and +finally without the field at all, so the model key alone can carry the target. Image model aliases: @@ -389,6 +432,11 @@ The generated binary is written to `dist/flow` (`dist/flow.exe` on Windows). record nor local managed file still exists. - **A retry might have reached Flow** — repeat it with the same idempotency key; do not create a new key for the same paid request. +- **1080p or 4K came back as 720p** — read the `note` in the response. Flow + rejected or failed the upsample pass; the 720p original is still delivered. + If Flow changed the wire format, override `VIDEO_UPSAMPLER_*_MODEL` or + `VIDEO_UPSAMPLE_ENUM_*` and retry. `python -m flow_server.sniff` captures the + Flow UI's own upsample request for comparison. - **MCP tools appear but fail** — verify `flow status`, then restart the MCP client after correcting its command/PATH configuration. diff --git a/flow-agent/flow_engine/__init__.py b/flow-agent/flow_engine/__init__.py index 30efdf1..40ed0e8 100644 --- a/flow-agent/flow_engine/__init__.py +++ b/flow-agent/flow_engine/__init__.py @@ -30,12 +30,16 @@ # ─── Public API ────────────────────────────────────────────── from .bridge import ExtensionBridge -from .config import ASPECTS, DEFAULT_PROJECT, ENDPOINTS, CLIENT_CTX, API_KEY, API_BASE +from .config import ( + ASPECTS, DEFAULT_PROJECT, ENDPOINTS, CLIENT_CTX, API_KEY, API_BASE, + NATIVE_VIDEO_RESOLUTION, VIDEO_UPSAMPLE_MODELS, CREDITS_PER_UPSAMPLE, +) from .generators import ( generate_video, edit_video, upload_image, generate_video_i2v, + upsample_video, poll_status, download_video, build_client_context, @@ -51,6 +55,7 @@ "edit_video", "upload_image", "generate_video_i2v", + "upsample_video", "poll_status", "download_video", "build_client_context", @@ -63,6 +68,9 @@ "CLIENT_CTX", "API_KEY", "API_BASE", + "NATIVE_VIDEO_RESOLUTION", + "VIDEO_UPSAMPLE_MODELS", + "CREDITS_PER_UPSAMPLE", # Media store "media_store", ] diff --git a/flow-agent/flow_engine/config.py b/flow-agent/flow_engine/config.py index d07997f..39cc61a 100644 --- a/flow-agent/flow_engine/config.py +++ b/flow-agent/flow_engine/config.py @@ -117,6 +117,7 @@ def _flow_binary_dir() -> str: "generate_r2v": "/v1/video:batchAsyncGenerateVideoReferenceImages", "generate_edit": "/v1/video:batchAsyncGenerateVideoEditVideo", "upload_image": "/v1/flow/uploadImage", + "upsample_video": "/v1/video:batchAsyncGenerateVideoUpsampleVideo", "poll_status": "/v1/video:batchCheckAsyncVideoGenerationStatus", "get_media": "/v1/media/{media_id}", "get_credits": "/v1/credits", @@ -132,6 +133,26 @@ def _flow_binary_dir() -> str: "edit": "abra_edit", } +# ─── Video upsampling (native generation is 720p) ───────────── +# Flow generates video at 720p and reaches 1080p/4K through a second +# "upsampler" pass on the finished media, exactly like the Flow UI's +# high-resolution download. Model keys and the resolution enum are part of an +# undocumented API, so both stay overridable without a code change. +NATIVE_VIDEO_RESOLUTION = "720p" + +VIDEO_UPSAMPLE_MODELS = { + "1080p": os.environ.get("VIDEO_UPSAMPLER_1080P_MODEL", "veo_3_1_upsampler_1080p"), + "4k": os.environ.get("VIDEO_UPSAMPLER_4K_MODEL", "veo_3_1_upsampler_4k"), +} + +VIDEO_UPSAMPLE_RESOLUTIONS = { + "1080p": os.environ.get("VIDEO_UPSAMPLE_ENUM_1080P", "VIDEO_RESOLUTION_1080P"), + "4k": os.environ.get("VIDEO_UPSAMPLE_ENUM_4K", "VIDEO_RESOLUTION_4K"), +} + +# 1080p upsampling is free; 4K is a paid, higher-tier operation. +CREDITS_PER_UPSAMPLE = {"1080p": 0, "4k": 50} + DURATIONS = [4, 6, 8, 10] DEFAULT_DURATION = 10 MAX_COUNT = 4 diff --git a/flow-agent/flow_engine/generators/__init__.py b/flow-agent/flow_engine/generators/__init__.py index b581362..0db9953 100644 --- a/flow-agent/flow_engine/generators/__init__.py +++ b/flow-agent/flow_engine/generators/__init__.py @@ -4,6 +4,7 @@ from .v2v import edit_video from .i2v import upload_image, generate_video_i2v from .t2i import generate_image, download_image, IMAGE_ASPECTS +from .upsample import upsample_video, normalise_resolution from .common import poll_status, download_video, build_client_context __all__ = [ @@ -12,6 +13,8 @@ "upload_image", "generate_video_i2v", "generate_image", + "upsample_video", + "normalise_resolution", "download_image", "IMAGE_ASPECTS", "poll_status", diff --git a/flow-agent/flow_engine/generators/upsample.py b/flow-agent/flow_engine/generators/upsample.py new file mode 100644 index 0000000..11cc516 --- /dev/null +++ b/flow-agent/flow_engine/generators/upsample.py @@ -0,0 +1,209 @@ +"""Flow Engine — Video upsampler (720p -> 1080p / 4K). + +Flow generates video at 720p. The higher-resolution download offered in the +Flow UI is a second pass over the finished media against an upsampler model, +submitted asynchronously and polled with the ordinary video status endpoint. + +The request shape below is part of Google's undocumented Flow API. The +resolution enum in particular is not published, so a rejected enum is retried +with the known spelling variants and finally omitted entirely — the upsampler +model key already encodes the target resolution. A rejected request never +starts a generation, so the ladder cannot double-charge. +""" + +import logging + +from ..config import ( + ENDPOINTS, + VIDEO_UPSAMPLE_MODELS, + VIDEO_UPSAMPLE_RESOLUTIONS, +) +from .common import build_client_context, build_generation_context, resolve_seed + +log = logging.getLogger("flow_engine.generators.upsample") + +# Ordered spelling candidates per tier. The configured value is always tried +# first; ``None`` means "send no resolution field at all". +_RESOLUTION_CANDIDATES = { + "1080p": ("VIDEO_RESOLUTION_1080P", "VIDEO_RESOLUTION_1080p", None), + "4k": ("VIDEO_RESOLUTION_4K", "VIDEO_RESOLUTION_4k", None), +} + +# Substrings that identify a request rejected for its shape rather than for a +# real generation failure (bad media ID, no credits, unusual activity...). +_SCHEMA_REJECTION_HINTS = ( + "resolution", + "invalid value", + "invalid_argument", + "unknown name", + "cannot find field", +) + + +def normalise_resolution(resolution) -> str | None: + """Map a user-facing resolution to an upsample tier key, or None. + + Returns None for the native 720p output, which needs no upsample pass. + """ + key = str(resolution or "").strip().lower().replace(" ", "") + if not key or key in {"720p", "720", "native", "source", "original"}: + return None + aliases = { + "1080": "1080p", + "1080p": "1080p", + "fhd": "1080p", + "hd": "1080p", + "full_hd": "1080p", + "fullhd": "1080p", + "4k": "4k", + "2160": "4k", + "2160p": "4k", + "uhd": "4k", + } + tier = aliases.get(key) + if tier is None: + raise ValueError( + f"Unsupported video resolution {resolution!r}; use '720p', '1080p', or '4k'." + ) + return tier + + +def _resolution_candidates(tier: str) -> list[str | None]: + configured = VIDEO_UPSAMPLE_RESOLUTIONS.get(tier) + ordered: list[str | None] = [configured] if configured else [] + for candidate in _RESOLUTION_CANDIDATES.get(tier, ()): + if candidate not in ordered: + ordered.append(candidate) + return ordered + + +def _error_message(result: dict) -> str: + data = result.get("data", {}) + reason = "" + message = result.get("error", "Unknown") + if isinstance(data, dict): + error = data.get("error", {}) + if isinstance(error, dict): + message = error.get("message", message) + for detail in error.get("details", []) or []: + if isinstance(detail, dict) and "reason" in detail: + reason = f" ({detail['reason']})" + break + elif data: + message = data + return f"{message}{reason}" + + +def _looks_like_schema_rejection(status: int, message: str) -> bool: + if status not in (400, 404): + return False + lowered = message.lower() + return any(hint in lowered for hint in _SCHEMA_REJECTION_HINTS) + + +def _parse_media_ids(data: dict) -> list[str]: + """Collect upsampled media IDs from either response shape.""" + if not isinstance(data, dict): + return [] + + ids: list[str] = [] + + def _add(value): + if isinstance(value, str) and value and value not in ids: + ids.append(value) + + for item in data.get("media", []) or []: + if isinstance(item, dict): + _add(item.get("name") or item.get("mediaId") or item.get("id")) + + # Older/legacy responses nest the new media under operations. + for item in data.get("operations", []) or []: + if not isinstance(item, dict): + continue + _add(item.get("name") or item.get("mediaId")) + nested = item.get("media") or item.get("operation") + if isinstance(nested, dict): + _add(nested.get("name") or nested.get("mediaId")) + + return ids + + +async def upsample_video( + bridge, + media_id: str, + aspect: str, + project_id: str, + resolution: str = "1080p", + seed: int = None, + scene_id: str = None, +) -> list[str] | None: + """Submit an upsample pass for one finished video. Returns new media IDs. + + ``resolution`` accepts '1080p' or '4k'. Poll the returned IDs with + ``flow_engine.generators.common.poll_status`` and download them the same way + as any generated video. + """ + tier = normalise_resolution(resolution) + if tier is None: + raise ValueError( + "Upsampling to 720p is a no-op; Flow already generates at 720p." + ) + + model_key = VIDEO_UPSAMPLE_MODELS.get(tier) + if not model_key: + raise ValueError(f"No upsampler model configured for {tier}.") + + last_error = None + for attempt, resolution_enum in enumerate(_resolution_candidates(tier)): + request_item = { + "aspectRatio": aspect, + "videoModelKey": model_key, + "seed": resolve_seed(seed), + "metadata": {"sceneId": scene_id} if scene_id else {}, + "videoInput": {"mediaId": media_id}, + } + if resolution_enum is not None: + request_item["resolution"] = resolution_enum + + body = { + "mediaGenerationContext": build_generation_context(), + "clientContext": build_client_context(project_id), + "requests": [request_item], + } + + log.info( + "Upsampling %s to %s [%s]%s", + media_id[:12], + tier, + model_key, + "" if resolution_enum is None else f" resolution={resolution_enum}", + ) + result = await bridge.api_request( + ENDPOINTS["upsample_video"], body, captcha_action="VIDEO_GENERATION" + ) + + status = result.get("status", 0) + if status == 200: + media_ids = _parse_media_ids(result.get("data", {})) + if not media_ids: + log.error("Upsample accepted but returned no media: %r", result.get("data")) + return None + log.info("Upsample submitted! %s -> %s", media_id[:12], ", ".join(media_ids)) + return media_ids + + last_error = _error_message(result) + if _looks_like_schema_rejection(status, last_error) and attempt + 1 < len( + _resolution_candidates(tier) + ): + log.warning( + "Upsample rejected the request shape (%s): %s — retrying with the " + "next resolution spelling.", + status, + last_error, + ) + continue + + log.error("Upsample failed (%s): %s", status, last_error) + raise ValueError(last_error) + + raise ValueError(last_error or "Upsample request was rejected.") diff --git a/flow-agent/flow_server/mcp/executor.py b/flow-agent/flow_server/mcp/executor.py index 09e12c3..4d14119 100644 --- a/flow-agent/flow_server/mcp/executor.py +++ b/flow-agent/flow_server/mcp/executor.py @@ -19,15 +19,28 @@ from flow_server.config import OUTPUT_DIR, _normalise_model from flow_server.media_types import extension_for_media, sniff_media_type -from flow_server.models import ImageGenerationRequest, VideoGenerationRequest +from flow_server.models import ImageGenerationRequest, VideoGenerationRequest, VideoUpsampleRequest from flow_server.mcp.helpers import _mcp_error, _mcp_upload_local_file, _mcp_download_url from flow_server.routes.system import health, list_models, get_flow_credits from flow_server.routes.media import get_history -from flow_server.routes.generation import openai_generate_image, openai_generate_video +from flow_server.routes.generation import ( + openai_generate_image, + openai_generate_video, + openai_upsample_video, +) log = logging.getLogger("flow_engine.openai_api") +def _resolution_summary(data) -> str: + """Summarise the delivered resolutions, so a client sees 1080p landed.""" + resolutions = [item.get("resolution") for item in data if item.get("resolution")] + if not resolutions: + return "" + ordered = sorted(set(resolutions), key=resolutions.index) + return " Resolution: " + ", ".join(ordered) + "." + + async def execute_mcp_tool(request_id, tool_name, arguments): text = "" images_b64 = [] @@ -168,6 +181,7 @@ async def execute_mcp_tool(request_id, tool_name, arguments): start_media_id=start_media_id, ref_media_ids=(list(arguments.get("ref_media_ids") or [])[:10] or None), is_video=is_edit, + resolution=arguments.get("resolution"), ) res = await openai_generate_video(req, x_client_id=None) @@ -179,8 +193,36 @@ async def execute_mcp_tool(request_id, tool_name, arguments): media_ids = [item.get("media_id") for item in data if item.get("media_id")] verb = "Edited" if is_edit else "Generated" text = (f"Success! {verb} {len(data)} video(s)." + + _resolution_summary(data) + ("\nURLs:\n" + "\n".join(urls) if urls else "") + ("\nMedia IDs: " + ", ".join(media_ids) if media_ids else "")) + if res.get("note"): + text += "\nNote: " + res["note"] + + elif tool_name == "upsample_flow_video": + media_id = str(arguments.get("media_id") or "").strip() + if not media_id: + return _mcp_error(request_id, -32602, "Error: 'media_id' is required.") + + seed = arguments.get("seed") + req = VideoUpsampleRequest( + media_id=media_id, + resolution=str(arguments.get("resolution") or "1080p"), + aspect=arguments.get("aspect", "landscape"), + seed=int(seed) if seed is not None else None, + ) + + res = await openai_upsample_video(req, x_client_id=None) + data = res.get("data", []) + if not data: + text = "No upsampled video returned by Flow Agent." + else: + item = data[0] + text = (f"Success! Upsampled to {item.get('resolution') or req.resolution}." + + (f"\nURL: {item['url']}" if item.get("url") else "") + + (f"\nMedia ID: {item['media_id']}" if item.get("media_id") else "") + + (f"\nUpsampled from: {item['source_media_id']}" + if item.get("source_media_id") else "")) elif tool_name == "upload_flow_media": file_path = arguments.get("file_path") diff --git a/flow-agent/flow_server/mcp/tools.py b/flow-agent/flow_server/mcp/tools.py index 147cabf..0722b28 100644 --- a/flow-agent/flow_server/mcp/tools.py +++ b/flow-agent/flow_server/mcp/tools.py @@ -90,7 +90,7 @@ def get_mcp_tools_list(): }, { "name": "generate_flow_video", - "description": "Generate 1-20 Flow videos with duration, aspect, start asset, and reference-media control.", + "description": "Generate 1-20 Flow videos with duration, aspect, start asset, reference-media, and delivery-resolution control (720p native, 1080p or 4K via Flow's upsampler).", "inputSchema": { "type": "object", "properties": { @@ -116,11 +116,36 @@ def get_mcp_tools_list(): "items": {"type": "string"}, "maxItems": 10, "description": "Optional Flow reference-media IDs for reference-to-video" + }, + "resolution": { + "type": "string", + "enum": ["720p", "1080p", "4k"], + "default": "720p", + "description": "Delivery resolution. Flow generates at 720p; '1080p' (free) or '4k' (paid, higher tier) add Flow's upsampler pass and the high-resolution file is returned first." } }, "required": ["prompt"] } }, + { + "name": "upsample_flow_video", + "description": "Upsample an existing Flow video to 1080p or 4K — the same high-resolution pass behind the Flow UI's HD download. Accepts a media ID or a local video path already in Flow history.", + "inputSchema": { + "type": "object", + "properties": { + "media_id": {"type": "string", "description": "Media ID of a finished Flow video, or a local video path/filename tracked in history"}, + "resolution": { + "type": "string", + "enum": ["1080p", "4k"], + "default": "1080p", + "description": "Target resolution: '1080p' is free, '4k' costs credits and needs a higher Flow tier" + }, + "aspect": {"type": "string", "enum": ["landscape", "portrait"], "default": "landscape", "description": "Aspect ratio of the source video"}, + "seed": {"type": "integer", "minimum": 0, "maximum": 4294967295, "description": "Optional explicit upsampler seed"} + }, + "required": ["media_id"] + } + }, { "name": "upload_flow_media", "description": "Upload an image or video to Google Flow from a local path, a public URL, or base64 data, and return its media ID.", diff --git a/flow-agent/flow_server/mcp_server.py b/flow-agent/flow_server/mcp_server.py index 2549473..705c616 100755 --- a/flow-agent/flow_server/mcp_server.py +++ b/flow-agent/flow_server/mcp_server.py @@ -252,7 +252,7 @@ def handle_tools_list(request_id): }, { "name": "generate_flow_video", - "description": "Generate 1-20 Flow videos with duration, aspect, start asset, seed, first-last frame, and reference-media control.", + "description": "Generate 1-20 Flow videos with duration, aspect, start asset, seed, first-last frame, reference-media, and delivery-resolution control (720p native, 1080p or 4K via Flow's upsampler).", "inputSchema": { "type": "object", "properties": { @@ -296,11 +296,36 @@ def handle_tools_list(request_id): "video_model": { "type": "string", "description": "Override the Flow videoModelKey (defaults to abra_t2v_s)" + }, + "resolution": { + "type": "string", + "enum": ["720p", "1080p", "4k"], + "default": "720p", + "description": "Delivery resolution. Flow generates at 720p; '1080p' (free) or '4k' (paid, higher tier) add Flow's upsampler pass and the high-resolution file is returned first." } }, "required": ["prompt"] } }, + { + "name": "upsample_flow_video", + "description": "Upsample an existing Flow video to 1080p or 4K — the same high-resolution pass behind the Flow UI's HD download. Accepts a media ID or a local video path already in Flow history.", + "inputSchema": { + "type": "object", + "properties": { + "media_id": {"type": "string", "description": "Media ID of a finished Flow video, or a local video path/filename tracked in history"}, + "resolution": { + "type": "string", + "enum": ["1080p", "4k"], + "default": "1080p", + "description": "Target resolution: '1080p' is free, '4k' costs credits and needs a higher Flow tier" + }, + "aspect": {"type": "string", "enum": ["landscape", "portrait"], "default": "landscape", "description": "Aspect ratio of the source video"}, + "seed": {"type": "integer", "minimum": 0, "maximum": 4294967295, "description": "Optional explicit upsampler seed"} + }, + "required": ["media_id"] + } + }, { "name": "generate_flow_sequence", "description": "Generate a continuity-chained run of shots: each shot starts on the previous shot's final frame, so cuts land on matching pixels. Returns clip paths in order. Requires FFmpeg.", @@ -336,6 +361,12 @@ def handle_tools_list(request_id): "ref_media_ids": { "type": "array", "items": {"type": "string"}, "maxItems": 10, "description": "Reference media applied to every shot, for style or character carry-through" + }, + "resolution": { + "type": "string", + "enum": ["720p", "1080p", "4k"], + "default": "720p", + "description": "Delivery resolution for every shot in the sequence" } }, "required": ["shots"] @@ -585,7 +616,8 @@ def call_generate_flow_image(prompt, size="1280x720", count=1, ref_image_path=No def call_generate_flow_video(prompt, aspect="landscape", start_image_path=None, duration=8, count=1, start_media_id=None, ref_media_ids=None, is_video=False, - seed=None, end_image_path=None, end_media_id=None, video_model=None): + seed=None, end_image_path=None, end_media_id=None, video_model=None, + resolution=None): if not prompt or not str(prompt).strip(): return "Error: 'prompt' is required and cannot be empty." prompt = str(prompt).strip() @@ -605,6 +637,8 @@ def call_generate_flow_video(prompt, aspect="landscape", start_image_path=None, payload["seed"] = int(seed) if video_model: payload["video_model"] = video_model + if resolution: + payload["resolution"] = resolution if end_media_id: payload["end_media_id"] = end_media_id @@ -643,9 +677,16 @@ def call_generate_flow_video(prompt, aspect="landscape", start_image_path=None, return "No videos returned by Flow Agent." urls = [item.get("url") for item in data if item.get("url")] media_ids = [item.get("media_id") for item in data if item.get("media_id")] - return (f"Success! Generated {len(data)} video(s)." + resolutions = [item.get("resolution") for item in data if item.get("resolution")] + summary = "" + if resolutions: + ordered = sorted(set(resolutions), key=resolutions.index) + summary = " Resolution: " + ", ".join(ordered) + "." + note = res_data.get("note") + return (f"Success! Generated {len(data)} video(s)." + summary + ("\nURLs:\n" + "\n".join(urls) if urls else "") - + ("\nMedia IDs: " + ", ".join(media_ids) if media_ids else "")) + + ("\nMedia IDs: " + ", ".join(media_ids) if media_ids else "") + + (f"\nNote: {note}" if note else "")) except urllib.error.HTTPError as e: try: err_msg = e.read().decode('utf-8') @@ -655,6 +696,49 @@ def call_generate_flow_video(prompt, aspect="landscape", start_image_path=None, except Exception as e: return f"Failed to communicate with Flow Agent server: {str(e)}" +def call_upsample_flow_video(media_id, resolution="1080p", aspect="landscape", seed=None): + """Upsample a finished Flow video to 1080p or 4K through the backend.""" + if not media_id or not str(media_id).strip(): + return "Error: 'media_id' is required." + payload = { + "media_id": str(media_id).strip(), + "resolution": str(resolution or "1080p"), + "aspect": aspect, + } + if seed is not None: + payload["seed"] = int(seed) + + try: + log_debug(f"Requesting {payload['resolution']} upsample for {payload['media_id']}") + req = urllib.request.Request( + f"{FLOW_API_URL}/v1/videos/upsample", + data=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(req, timeout=900) as response: + if response.status != 200: + return f"Error upsampling video ({response.status})" + res_data = json.loads(response.read().decode("utf-8")) + data = res_data.get("data", []) + if not data: + return "No upsampled video returned by Flow Agent." + item = data[0] + return (f"Success! Upsampled to {item.get('resolution') or payload['resolution']}." + + (f"\nURL: {item['url']}" if item.get("url") else "") + + (f"\nMedia ID: {item['media_id']}" if item.get("media_id") else "") + + (f"\nUpsampled from: {item['source_media_id']}" + if item.get("source_media_id") else "")) + except urllib.error.HTTPError as e: + try: + err_msg = e.read().decode("utf-8") + except Exception: + err_msg = str(e) + return f"Error upsampling video ({e.code}): {err_msg}" + except Exception as e: + return f"Failed to communicate with Flow Agent server: {str(e)}" + + def call_edit_flow_video(prompt, media_id=None, video_path=None, aspect="landscape", duration=8, ref_media_ids=None): if not media_id and video_path: @@ -841,7 +925,7 @@ def _post_video_json(payload, timeout=900): def call_generate_flow_sequence(shots, aspect="landscape", duration=8, output_dir=None, start_image_path=None, seed=None, video_model=None, - ref_media_ids=None): + ref_media_ids=None, resolution=None): """Generate a continuity-chained run of shots. Each shot after the first starts on the previous shot's final frame, so cuts @@ -879,6 +963,8 @@ def call_generate_flow_sequence(shots, aspect="landscape", duration=8, output_di payload["video_model"] = video_model if ref_media_ids: payload["ref_media_ids"] = list(ref_media_ids)[:10] + if resolution: + payload["resolution"] = resolution if carry_frame: try: payload["image_base64"] = _file_data_uri(carry_frame) @@ -1020,6 +1106,15 @@ def handle_tool_call(request_id, tool_name, arguments): arguments.get("end_image_path"), arguments.get("end_media_id"), arguments.get("video_model"), + arguments.get("resolution"), + ) + content = [{"type": "text", "text": text}] + elif tool_name == "upsample_flow_video": + text = call_upsample_flow_video( + arguments.get("media_id"), + arguments.get("resolution", "1080p"), + arguments.get("aspect", "landscape"), + arguments.get("seed"), ) content = [{"type": "text", "text": text}] elif tool_name == "generate_flow_sequence": @@ -1032,6 +1127,7 @@ def handle_tool_call(request_id, tool_name, arguments): arguments.get("seed"), arguments.get("video_model"), arguments.get("ref_media_ids"), + arguments.get("resolution"), ) content = [{"type": "text", "text": json.dumps(result, indent=2)}] elif tool_name == "extract_video_frame": diff --git a/flow-agent/flow_server/models.py b/flow-agent/flow_server/models.py index 5b6a0a5..9eea3cb 100644 --- a/flow-agent/flow_server/models.py +++ b/flow-agent/flow_server/models.py @@ -31,12 +31,31 @@ class VideoGenerationRequest(BaseModel): is_video: Optional[bool] = Field(False, description="True if the pre-uploaded reference is a video") seed: Optional[int] = Field(None, ge=0, le=4294967295, description="Optional explicit generation seed") video_model: Optional[str] = Field(None, description="Optional Flow videoModelKey override") + resolution: Optional[str] = Field( + None, + description=( + "Optional delivery resolution: '720p' (native, default), '1080p', or '4k'. " + "Above 720p the finished video is run through Flow's upsampler pass, " + "the same step behind the Flow UI's high-resolution download." + ), + ) + + +class VideoUpsampleRequest(BaseModel): + """Upsample one already-generated Flow video to 1080p or 4K.""" + + media_id: str = Field(..., description="Media ID of a finished Flow video, or a local video path/filename in history") + resolution: str = Field("1080p", description="Target resolution: '1080p' (free) or '4k' (paid, higher tier)") + aspect: str = Field("landscape", description="Aspect ratio of the source video (portrait or landscape)") + seed: Optional[int] = Field(None, ge=0, le=4294967295, description="Optional explicit upsampler seed") class GeneratedMedia(BaseModel): url: Optional[str] = None media_id: Optional[str] = None warning: Optional[str] = None + resolution: Optional[str] = Field(None, description="Delivered resolution of this file") + source_media_id: Optional[str] = Field(None, description="Media ID this file was upsampled from") class VideoGenerationResult(BaseModel): diff --git a/flow-agent/flow_server/routes/generation.py b/flow-agent/flow_server/routes/generation.py index e7120bc..3d81da5 100644 --- a/flow-agent/flow_server/routes/generation.py +++ b/flow-agent/flow_server/routes/generation.py @@ -13,7 +13,12 @@ from fastapi.responses import JSONResponse from flow_server.config import OUTPUT_DIR, map_size_to_aspect -from flow_server.models import ImageGenerationRequest, VideoGenerationRequest, VideoGenerationResult +from flow_server.models import ( + ImageGenerationRequest, + VideoGenerationRequest, + VideoGenerationResult, + VideoUpsampleRequest, +) from flow_server.idempotency import get_idempotency_store from flow_server.jobs import get_job_store from flow_server.history import MediaNotFoundError @@ -23,8 +28,9 @@ from flow_engine import DEFAULT_PROJECT from flow_engine.bridge import target_client_id_var -from flow_engine.config import CREDITS_PER_VIDEO +from flow_engine.config import CREDITS_PER_UPSAMPLE, CREDITS_PER_VIDEO, NATIVE_VIDEO_RESOLUTION from flow_engine.generators.t2i import generate_image, download_image +from flow_engine.generators.upsample import normalise_resolution, upsample_video # Setup logging (format configured centrally in flow_engine/__init__.py, imported above) log = logging.getLogger("flow_engine.openai_api") @@ -410,12 +416,249 @@ async def get_video_generation(job_id: str): return job +async def _resolve_upsample_tier(resolution) -> Optional[str]: + """Validate a requested delivery resolution, returning the upsample tier.""" + try: + return normalise_resolution(resolution) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + +async def _ensure_upsample_credits(active_bridge, tier: str, count: int) -> None: + """Refuse a paid upsample the account cannot afford. 1080p is free.""" + cost_each = CREDITS_PER_UPSAMPLE.get(tier, 0) + if cost_each <= 0: + return + try: + cred_res = await active_bridge.api_request( + "/v1/credits", body=None, captcha_action=None, method="GET" + ) + cred_data = cred_res.get("data", cred_res) if isinstance(cred_res, dict) else {} + balance = int(cred_data.get("credits", 0)) + except Exception: + return + if balance < cost_each * count: + raise HTTPException( + status_code=402, + detail=( + f"Not enough credits: {balance} left, but {count} {tier} upsample(s) " + f"cost {cost_each} each." + ), + ) + + +async def _upsample_and_download( + active_bridge, + source_media_id: str, + aspect_key: str, + project_id: str, + tier: str, + *, + prompt: str, + seed: Optional[int] = None, + index: int = 0, +): + """Run one upsample pass end to end: submit, poll, download, record. + + Returns the new media entry, or raises ValueError with Flow's own message. + """ + media_ids = await upsample_video( + active_bridge, + source_media_id, + aspect_key, + project_id, + resolution=tier, + seed=seed, + ) + if not media_ids: + raise ValueError(f"Flow accepted the {tier} upsample but returned no media.") + + from flow_engine.generators.common import poll_status, download_video + + upsampled_id = media_ids[0] + if not await poll_status(active_bridge, upsampled_id, project_id): + raise ValueError(f"The {tier} upsample of {source_media_id} did not finish.") + + timestamp = int(time.time()) + filename = f"flow_vid_{timestamp}_{uuid.uuid4().hex[:6]}_{index + 1}_{tier}.mp4" + out_path = os.path.join(OUTPUT_DIR, filename) + if not await download_video(active_bridge, upsampled_id, out_path): + raise ValueError(f"Downloading the {tier} upsample of {upsampled_id} failed.") + + out_path = ensure_correct_extension(out_path) + filename = os.path.basename(out_path) + served_url, r2_key = await publish(filename, out_path) + await append_to_history( + "video", + served_url, + prompt, + upsampled_id, + r2_key, + local_path=out_path, + project_id=project_id, + ) + return { + "url": served_url, + "media_id": upsampled_id, + "resolution": tier, + "source_media_id": source_media_id, + } + + +@router.post( + "/v1/videos/upsample", + dependencies=[Depends(verify_api_key)], + response_model=VideoGenerationResult, + responses={202: {"model": VideoGenerationResult, "description": "Upsample is already processing"}}, +) +async def openai_upsample_video( + req: VideoUpsampleRequest, + x_client_id: Optional[str] = Header(None, alias="X-Client-Id"), + idempotency_key: Optional[str] = Header(None, alias="Idempotency-Key"), +): + """Upsample an existing Flow video to 1080p or 4K and download the result. + + Flow generates video at 720p; this is the second pass that the Flow UI runs + behind its high-resolution download. + """ + x_client_id = _header_string(x_client_id) + key = _header_string(idempotency_key) + created = int(time.time()) + job_id = f"upsample_{uuid.uuid4().hex}" + idem_store = get_idempotency_store(OUTPUT_DIR) + job_store = get_job_store(OUTPUT_DIR) + + if key: + claim = await idem_store.claim( + key, + _request_payload(req, "upsample", x_client_id), + job_id=job_id, + created=created, + ) + if claim.action == "conflict": + raise HTTPException( + status_code=409, + detail="Idempotency-Key was already used with a different upsample request.", + ) + if claim.action == "replay": + return claim.record["response"] + if claim.action == "failed": + error = claim.record.get("error", {}) + raise HTTPException( + status_code=int(error.get("status_code", 500)), + detail=error.get("detail", "The original idempotent upsample failed."), + ) + if claim.action == "processing": + existing_job_id = claim.record.get("job_id") + existing_job = await job_store.get(existing_job_id) if existing_job_id else None + if existing_job and existing_job.get("status") == "succeeded": + await idem_store.succeed(key, existing_job) + return existing_job + if existing_job and existing_job.get("status") == "failed": + error = existing_job.get("error", {}) + await idem_store.fail( + key, + int(error.get("status_code", 500)), + error.get("detail", "The original upsample failed."), + ) + raise HTTPException( + status_code=int(error.get("status_code", 500)), + detail=error.get("detail", "The original upsample failed."), + ) + pending = { + "job_id": existing_job_id, + "status": "processing", + "created": claim.record.get("created") or created, + "data": [], + } + if existing_job_id and not existing_job: + await job_store.put(existing_job_id, pending) + return JSONResponse(status_code=202, content=pending) + + pending = {"job_id": job_id, "status": "processing", "created": created, "data": []} + await job_store.put(job_id, pending) + + try: + generated = await _upsample_video(req, x_client_id) + except HTTPException as exc: + error = {"status_code": exc.status_code, "detail": str(exc.detail)} + await job_store.update(job_id, status="failed", error=error) + if key: + await idem_store.fail(key, exc.status_code, str(exc.detail)) + raise + except Exception as exc: + error = {"status_code": 500, "detail": str(exc)} + await job_store.update(job_id, status="failed", error=error) + if key: + await idem_store.fail(key, 500, str(exc)) + raise + + result = dict(generated) + result.update(job_id=job_id, status="succeeded") + await job_store.put(job_id, result) + if key: + await idem_store.succeed(key, result) + return result + + +async def _upsample_video(req: VideoUpsampleRequest, x_client_id: Optional[str] = None): + """Upsample one finished video to a higher delivery resolution.""" + target_client_id_var.set(x_client_id) + active_bridge = await get_active_bridge() + project_id = os.environ.get("DEFAULT_PROJECT", DEFAULT_PROJECT) + + tier = await _resolve_upsample_tier(req.resolution) + if tier is None: + raise HTTPException( + status_code=400, + detail=( + f"Flow already delivers {NATIVE_VIDEO_RESOLUTION} natively; " + "request '1080p' or '4k' to upsample." + ), + ) + + from flow_engine import ASPECTS + aspect_key = ASPECTS.get(req.aspect, "VIDEO_ASPECT_RATIO_PORTRAIT") + + try: + source_media_id = await resolve_media_reference( + req.media_id, + active_bridge, + expected_type="video", + project_id=project_id, + ) + except MediaNotFoundError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + except (ValueError, RuntimeError) as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + await _ensure_upsample_credits(active_bridge, tier, 1) + + try: + entry = await _upsample_and_download( + active_bridge, + source_media_id, + aspect_key, + project_id, + tier, + prompt=f"Upsampled to {tier}", + seed=req.seed, + ) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + return {"created": int(time.time()), "data": [entry]} + + async def _generate_video(req: VideoGenerationRequest, x_client_id: Optional[str] = None): """Generate videos from a prompt (and optional start image).""" target_client_id_var.set(x_client_id) active_bridge = await get_active_bridge() project_id = os.environ.get("DEFAULT_PROJECT", DEFAULT_PROJECT) + # Reject an unusable resolution before anything is paid for. + upsample_tier = await _resolve_upsample_tier(req.resolution) + # Credit gate: only allow as many videos as the balance can afford. requested_n = req.n generation_count = requested_n @@ -624,7 +867,11 @@ async def poll_and_download(media_id: str, index: int): for r in results: if r: - data_outputs.append({"url": r["url"], "media_id": r.get("media_id")}) + data_outputs.append({ + "url": r["url"], + "media_id": r.get("media_id"), + "resolution": NATIVE_VIDEO_RESOLUTION, + }) await append_to_history( "video", r["url"], @@ -638,13 +885,58 @@ async def poll_and_download(media_id: str, index: int): if not data_outputs: raise HTTPException(status_code=500, detail="Failed to complete video generations or downloads.") - resp = { - "created": timestamp, - "data": data_outputs - } + notes = [] if requested_n != len(data_outputs): - resp["note"] = ( + notes.append( f"Requested {requested_n} video(s); generated {len(data_outputs)} " f"(each {req.duration}s video costs {cost_each} credits)." ) + + # Flow generates at 720p. A higher delivery resolution is a second + # upsampler pass over the finished media, so it runs only once every clip + # has landed. The 720p originals stay in history and in the response. + if upsample_tier: + await _ensure_upsample_credits(active_bridge, upsample_tier, len(data_outputs)) + + async def upsample_entry(entry, index): + source_id = entry.get("media_id") + if not source_id: + return None, "a generated clip had no media ID to upsample" + try: + return await _upsample_and_download( + active_bridge, + source_id, + aspect_key, + project_id, + upsample_tier, + prompt=req.prompt, + seed=req.seed, + index=index, + ), None + except (ValueError, RuntimeError) as exc: + log.error("Upsample to %s failed for %s: %s", upsample_tier, source_id, exc) + return None, str(exc) + + upsampled = await asyncio.gather( + *(upsample_entry(entry, i) for i, entry in enumerate(data_outputs)) + ) + high_res = [entry for entry, _ in upsampled if entry] + failures = [reason for entry, reason in upsampled if not entry and reason] + + if high_res: + # Lead with the upsampled files so clients that consume data[0] as + # "the" output write the high-resolution video. + data_outputs = high_res + data_outputs + if failures: + notes.append( + f"{len(failures)} clip(s) could not be upsampled to {upsample_tier} and are " + f"returned at {NATIVE_VIDEO_RESOLUTION}: {failures[0]}" + ) + + resp = { + "created": timestamp, + "data": data_outputs + } + if notes: + resp["note"] = " ".join(notes) return resp diff --git a/flow-agent/main.py b/flow-agent/main.py index adf1665..f0d1a7d 100644 --- a/flow-agent/main.py +++ b/flow-agent/main.py @@ -413,6 +413,8 @@ def _save_outputs(result, output_path, media_label, *, explicit_output): ) if item.get("media_id"): print(f"media_id={item['media_id']}") + if item.get("resolution"): + print(f"resolution={item['resolution']}") saved.append(destination) if result.get("note"): print(result["note"]) @@ -556,6 +558,17 @@ def cmd_video(argv): parser.add_argument("--end", metavar="IMAGE", help="End image path or media ID (use with --start)") parser.add_argument("--ref", "-r", nargs="+", metavar="IMAGE", help="Reference image paths or media IDs") parser.add_argument("--project-id", "-p", help="Deprecated; configure DEFAULT_PROJECT on the backend") + parser.add_argument( + "--resolution", + "-R", + choices=["720p", "1080p", "4k"], + default="720p", + help=( + "Delivery resolution. Flow generates at 720p; 1080p (free) and 4k (paid, " + "higher tier) add Flow's upsampler pass and the high-resolution file is " + "written to --output." + ), + ) parser.add_argument("--idempotency-key", help="Reuse a previous paid request safely") args = parser.parse_args(argv) @@ -573,6 +586,8 @@ def cmd_video(argv): "duration": args.duration, "n": args.count, } + if args.resolution != "720p": + payload["resolution"] = args.resolution if args.edit: payload.update({"start_media_id": args.edit, "is_video": True}) elif args.start: @@ -674,6 +689,46 @@ def cmd_upload(argv): print(json.dumps({"path": os.path.abspath(path), **result}, indent=2)) +def cmd_upsample(argv): + parser = argparse.ArgumentParser( + prog="flow upsample", + description=( + "Upsample a finished Flow video to 1080p or 4k — the high-resolution " + "pass behind the Flow UI's HD download." + ), + ) + parser.add_argument("media_id", help="Flow video media ID, or a local video path tracked in history") + parser.add_argument( + "--resolution", + "-R", + choices=["1080p", "4k"], + default="1080p", + help="Target resolution (1080p is free; 4k costs credits and needs a higher tier)", + ) + parser.add_argument("--aspect", "-a", choices=["portrait", "landscape"], default="landscape") + parser.add_argument("--output", "-o", default=None, help="Exact output file path") + parser.add_argument("--seed", type=int, help="Optional explicit upsampler seed") + parser.add_argument("--idempotency-key", help="Reuse a previous paid request safely") + args = parser.parse_args(argv) + + _wait_for_generation_ready() + output_path, explicit_output = _requested_output_path( + args.output, f"video_{args.resolution}.mp4" + ) + payload = { + "media_id": args.media_id, + "resolution": args.resolution, + "aspect": args.aspect, + } + if args.seed is not None: + payload["seed"] = args.seed + + idempotency_key = args.idempotency_key or uuid.uuid4().hex + result = _post_generation("/v1/videos/upsample", payload, idempotency_key, timeout=900) + result = _poll_video_job(result, idempotency_key) + _save_outputs(result, output_path, "video", explicit_output=explicit_output) + + def cmd_credits(argv): parser = argparse.ArgumentParser(prog="flow credits", description="Show Flow credits.") parser.parse_args(argv) @@ -710,6 +765,7 @@ def cmd_batch(argv): "image": cmd_image, "batch": cmd_batch, "video": cmd_video, + "upsample": cmd_upsample, "edit": cmd_edit, "upload": cmd_upload, "credits": cmd_credits, @@ -727,6 +783,7 @@ def _usage(): print(" image Generate images") print(" batch Generate batch images in parallel") print(" video Generate videos") + print(" upsample Upsample a video to 1080p or 4k") print(" edit Edit a video") print(" upload Upload media") print(" credits Show Flow credits") diff --git a/flow-agent/tests/test_video_upsample.py b/flow-agent/tests/test_video_upsample.py new file mode 100644 index 0000000..4da7ee2 --- /dev/null +++ b/flow-agent/tests/test_video_upsample.py @@ -0,0 +1,355 @@ +import tempfile +import unittest +from unittest.mock import AsyncMock, patch + +from fastapi import HTTPException + +from flow_engine.config import ENDPOINTS +from flow_engine.generators import upsample as upsample_mod +from flow_engine.generators.upsample import normalise_resolution, upsample_video +from flow_server.idempotency import clear_idempotency_store_cache +from flow_server.jobs import clear_job_store_cache +from flow_server.models import VideoGenerationRequest, VideoUpsampleRequest +from flow_server.routes import generation + + +MP4 = b"\x00\x00\x00\x18ftypisom" + b"test-video" + + +class RecordingBridge: + """Bridge stub that records every upsample submission it is handed.""" + + def __init__(self, responses): + self.responses = list(responses) + self.calls = [] + + def _select_client_for_cost(self, _cost): + return None + + async def api_request(self, path, body=None, captcha_action=None, method="POST"): + if path == "/v1/credits": + return {"data": {"credits": 1000}} + self.calls.append({ + "path": path, + "body": body, + "captcha_action": captcha_action, + "method": method, + }) + return self.responses.pop(0) + + +def _ok(media_ids): + return {"status": 200, "data": {"media": [{"name": mid} for mid in media_ids]}} + + +def _rejected(message): + return { + "status": 400, + "data": {"error": {"message": message, "details": [{"reason": "INVALID_ARGUMENT"}]}}, + } + + +async def _publish(filename, _path): + return f"http://test/download/{filename}", None + + +async def _append_history(*_args, **_kwargs): + return None + + +async def _download(_bridge, _media_id, path): + with open(path, "wb") as handle: + handle.write(MP4) + return True + + +class ResolutionAliasTests(unittest.TestCase): + def test_native_and_higher_resolutions_map_to_tiers(self): + for native in (None, "", "720p", "720", "native", "original"): + self.assertIsNone(normalise_resolution(native)) + for value in ("1080p", "1080", "FHD", "Full HD"): + self.assertEqual(normalise_resolution(value), "1080p") + for value in ("4k", "4K", "2160p", "uhd"): + self.assertEqual(normalise_resolution(value), "4k") + + def test_unknown_resolution_is_rejected_before_any_request(self): + with self.assertRaises(ValueError) as exc: + normalise_resolution("8k") + self.assertIn("720p", str(exc.exception)) + + +class UpsampleRequestTests(unittest.IsolatedAsyncioTestCase): + async def test_request_targets_the_upsampler_endpoint_and_model(self): + bridge = RecordingBridge([_ok(["upsampled-1"])]) + + media_ids = await upsample_video( + bridge, + "source-media-1", + "VIDEO_ASPECT_RATIO_LANDSCAPE", + "project-1", + resolution="1080p", + seed=7, + ) + + self.assertEqual(media_ids, ["upsampled-1"]) + self.assertEqual(len(bridge.calls), 1) + call = bridge.calls[0] + self.assertEqual(call["path"], ENDPOINTS["upsample_video"]) + self.assertEqual(call["path"], "/v1/video:batchAsyncGenerateVideoUpsampleVideo") + self.assertEqual(call["captcha_action"], "VIDEO_GENERATION") + + item = call["body"]["requests"][0] + self.assertEqual(item["videoModelKey"], "veo_3_1_upsampler_1080p") + self.assertEqual(item["resolution"], "VIDEO_RESOLUTION_1080P") + self.assertEqual(item["videoInput"], {"mediaId": "source-media-1"}) + self.assertEqual(item["aspectRatio"], "VIDEO_ASPECT_RATIO_LANDSCAPE") + self.assertEqual(item["seed"], 7) + self.assertEqual( + call["body"]["clientContext"]["projectId"], "project-1" + ) + + async def test_4k_uses_the_4k_upsampler_model(self): + bridge = RecordingBridge([_ok(["upsampled-4k"])]) + + await upsample_video(bridge, "source", "VIDEO_ASPECT_RATIO_PORTRAIT", "p", resolution="4k") + + item = bridge.calls[0]["body"]["requests"][0] + self.assertEqual(item["videoModelKey"], "veo_3_1_upsampler_4k") + self.assertEqual(item["resolution"], "VIDEO_RESOLUTION_4K") + + async def test_rejected_resolution_enum_retries_the_next_spelling(self): + bridge = RecordingBridge([ + _rejected("Invalid value at 'requests[0].resolution' (TYPE_ENUM)"), + _ok(["upsampled-2"]), + ]) + + media_ids = await upsample_video(bridge, "source", "VIDEO_ASPECT_RATIO_LANDSCAPE", "p") + + self.assertEqual(media_ids, ["upsampled-2"]) + self.assertEqual(len(bridge.calls), 2) + self.assertEqual( + bridge.calls[0]["body"]["requests"][0]["resolution"], "VIDEO_RESOLUTION_1080P" + ) + self.assertEqual( + bridge.calls[1]["body"]["requests"][0]["resolution"], "VIDEO_RESOLUTION_1080p" + ) + + async def test_last_resort_attempt_omits_the_resolution_field(self): + bridge = RecordingBridge([ + _rejected("Invalid value at 'requests[0].resolution'"), + _rejected("Cannot find field: resolution"), + _ok(["upsampled-3"]), + ]) + + media_ids = await upsample_video(bridge, "source", "VIDEO_ASPECT_RATIO_LANDSCAPE", "p") + + self.assertEqual(media_ids, ["upsampled-3"]) + self.assertEqual(len(bridge.calls), 3) + # The model key alone encodes the target resolution. + self.assertNotIn("resolution", bridge.calls[2]["body"]["requests"][0]) + + async def test_a_real_failure_is_not_retried(self): + bridge = RecordingBridge([ + {"status": 429, "data": {"error": {"message": "UNUSUAL_ACTIVITY"}}}, + ]) + + with self.assertRaises(ValueError) as exc: + await upsample_video(bridge, "source", "VIDEO_ASPECT_RATIO_LANDSCAPE", "p") + + self.assertIn("UNUSUAL_ACTIVITY", str(exc.exception)) + self.assertEqual(len(bridge.calls), 1) + + async def test_upsampling_to_native_resolution_is_refused(self): + bridge = RecordingBridge([]) + with self.assertRaises(ValueError): + await upsample_video(bridge, "source", "a", "p", resolution="720p") + self.assertEqual(bridge.calls, []) + + async def test_media_ids_are_read_from_the_legacy_operations_shape(self): + bridge = RecordingBridge([ + {"status": 200, "data": {"operations": [{"media": {"name": "legacy-1"}}]}}, + ]) + + self.assertEqual( + await upsample_video(bridge, "source", "a", "p"), ["legacy-1"] + ) + + +class UpsampleRouteTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.tempdir = tempfile.TemporaryDirectory() + clear_idempotency_store_cache() + clear_job_store_cache() + self.output_patch = patch.object(generation, "OUTPUT_DIR", self.tempdir.name) + self.output_patch.start() + + def tearDown(self): + self.output_patch.stop() + clear_idempotency_store_cache() + clear_job_store_cache() + self.tempdir.cleanup() + + async def test_upsample_route_downloads_the_high_resolution_file_once(self): + bridge = RecordingBridge([_ok(["upsampled-1"])]) + request = VideoUpsampleRequest(media_id="source-media-1", resolution="1080p") + + with ( + patch.object(generation, "get_active_bridge", AsyncMock(return_value=bridge)), + patch.object(generation, "resolve_media_reference", AsyncMock(return_value="source-media-1")), + patch("flow_engine.generators.common.poll_status", AsyncMock(return_value=True)), + patch("flow_engine.generators.common.download_video", _download), + patch.object(generation, "publish", _publish), + patch.object(generation, "append_to_history", _append_history), + ): + first = await generation.openai_upsample_video(request, None, "upsample-key-1") + retry = await generation.openai_upsample_video(request, None, "upsample-key-1") + polled = await generation.get_video_generation(first["job_id"]) + + self.assertEqual(first, retry) + self.assertEqual(first, polled) + self.assertEqual(first["status"], "succeeded") + entry = first["data"][0] + self.assertEqual(entry["resolution"], "1080p") + self.assertEqual(entry["media_id"], "upsampled-1") + self.assertEqual(entry["source_media_id"], "source-media-1") + # The retry replayed the stored result instead of paying again. + self.assertEqual(len(bridge.calls), 1) + + async def test_a_bad_resolution_is_rejected_without_calling_flow(self): + bridge = RecordingBridge([]) + with ( + patch.object(generation, "get_active_bridge", AsyncMock(return_value=bridge)), + ): + with self.assertRaises(HTTPException) as exc: + await generation.openai_upsample_video( + VideoUpsampleRequest(media_id="m", resolution="8k"), None, None + ) + self.assertEqual(exc.exception.status_code, 400) + self.assertEqual(bridge.calls, []) + + async def test_video_generation_at_1080p_returns_the_upsampled_file_first(self): + bridge = RecordingBridge([_ok(["upsampled-1"])]) + request = VideoGenerationRequest(prompt="ocean waves", resolution="1080p") + + with ( + patch.object(generation, "get_active_bridge", AsyncMock(return_value=bridge)), + patch("flow_engine.generators.t2v.generate_video", AsyncMock(return_value=["native-1"])), + patch("flow_engine.generators.common.poll_status", AsyncMock(return_value=True)), + patch("flow_engine.generators.common.download_video", _download), + patch.object(generation, "publish", _publish), + patch.object(generation, "append_to_history", _append_history), + ): + result = await generation.openai_generate_video(request, None, None) + + data = result["data"] + self.assertEqual(len(data), 2) + self.assertEqual(data[0]["resolution"], "1080p") + self.assertEqual(data[0]["media_id"], "upsampled-1") + self.assertEqual(data[0]["source_media_id"], "native-1") + # The 720p original is still returned and still downloadable. + self.assertEqual(data[1]["resolution"], "720p") + self.assertEqual(data[1]["media_id"], "native-1") + + async def test_default_video_generation_stays_native_and_never_upsamples(self): + bridge = RecordingBridge([]) + never = AsyncMock(side_effect=AssertionError("upsample ran without being asked")) + + with ( + patch.object(generation, "get_active_bridge", AsyncMock(return_value=bridge)), + patch("flow_engine.generators.t2v.generate_video", AsyncMock(return_value=["native-1"])), + patch("flow_engine.generators.common.poll_status", AsyncMock(return_value=True)), + patch("flow_engine.generators.common.download_video", _download), + patch.object(generation, "publish", _publish), + patch.object(generation, "append_to_history", _append_history), + patch.object(generation, "upsample_video", never), + ): + result = await generation.openai_generate_video( + VideoGenerationRequest(prompt="ocean waves"), None, None + ) + + self.assertEqual([item["resolution"] for item in result["data"]], ["720p"]) + self.assertEqual(never.await_count, 0) + + async def test_a_failed_upsample_still_delivers_the_720p_video_with_a_note(self): + bridge = RecordingBridge([]) + failing = AsyncMock(side_effect=ValueError("upsampler is unavailable")) + + with ( + patch.object(generation, "get_active_bridge", AsyncMock(return_value=bridge)), + patch("flow_engine.generators.t2v.generate_video", AsyncMock(return_value=["native-1"])), + patch("flow_engine.generators.common.poll_status", AsyncMock(return_value=True)), + patch("flow_engine.generators.common.download_video", _download), + patch.object(generation, "publish", _publish), + patch.object(generation, "append_to_history", _append_history), + patch.object(generation, "upsample_video", failing), + ): + result = await generation.openai_generate_video( + VideoGenerationRequest(prompt="ocean waves", resolution="1080p"), None, None + ) + + self.assertEqual([item["resolution"] for item in result["data"]], ["720p"]) + self.assertIn("could not be upsampled", result["note"]) + self.assertIn("upsampler is unavailable", result["note"]) + + async def test_4k_is_refused_when_credits_cannot_cover_it(self): + class BrokeBridge(RecordingBridge): + async def api_request(self, path, body=None, captcha_action=None, method="POST"): + if path == "/v1/credits": + return {"data": {"credits": 10}} + return await super().api_request(path, body, captcha_action, method) + + bridge = BrokeBridge([]) + with ( + patch.object(generation, "get_active_bridge", AsyncMock(return_value=bridge)), + patch.object(generation, "resolve_media_reference", AsyncMock(return_value="source-1")), + ): + with self.assertRaises(HTTPException) as exc: + await generation.openai_upsample_video( + VideoUpsampleRequest(media_id="source-1", resolution="4k"), None, None + ) + + self.assertEqual(exc.exception.status_code, 402) + self.assertEqual(bridge.calls, []) + + +class UpsampleToolSurfaceTests(unittest.TestCase): + def test_both_mcp_transports_expose_the_same_upsample_tool(self): + from flow_server.mcp.tools import get_mcp_tools_list + from flow_server.mcp_server import handle_tools_list + + sse = {tool["name"]: tool for tool in get_mcp_tools_list()} + stdio = {tool["name"]: tool for tool in handle_tools_list(1)["result"]["tools"]} + + for surface in (sse, stdio): + self.assertIn("upsample_flow_video", surface) + tool = surface["upsample_flow_video"] + self.assertEqual( + tool["inputSchema"]["properties"]["resolution"]["enum"], ["1080p", "4k"] + ) + self.assertEqual(tool["inputSchema"]["required"], ["media_id"]) + + video = surface["generate_flow_video"]["inputSchema"]["properties"] + self.assertEqual(video["resolution"]["enum"], ["720p", "1080p", "4k"]) + self.assertEqual(video["resolution"]["default"], "720p") + + +class UpsampleConfigTests(unittest.IsolatedAsyncioTestCase): + async def test_model_key_and_resolution_enum_come_from_configuration(self): + # The Flow upsample API is undocumented, so a wire-format change must be + # fixable from configuration (VIDEO_UPSAMPLER_*_MODEL / + # VIDEO_UPSAMPLE_ENUM_*) rather than a code edit. + bridge = RecordingBridge([_ok(["upsampled-1"])]) + + with ( + patch.dict(upsample_mod.VIDEO_UPSAMPLE_MODELS, {"1080p": "custom_upsampler"}), + patch.dict(upsample_mod.VIDEO_UPSAMPLE_RESOLUTIONS, {"1080p": "CUSTOM_ENUM"}), + ): + await upsample_video(bridge, "source", "VIDEO_ASPECT_RATIO_LANDSCAPE", "p") + + item = bridge.calls[0]["body"]["requests"][0] + self.assertEqual(item["videoModelKey"], "custom_upsampler") + self.assertEqual(item["resolution"], "CUSTOM_ENUM") + + +if __name__ == "__main__": + unittest.main() diff --git a/flow-agent/uv.lock b/flow-agent/uv.lock index 60056b5..8506bcc 100644 --- a/flow-agent/uv.lock +++ b/flow-agent/uv.lock @@ -245,7 +245,7 @@ name = "exceptiongroup" version = "1.3.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "typing-extensions", marker = "python_full_version < '3.13'" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" } wheels = [ @@ -270,7 +270,7 @@ wheels = [ [[package]] name = "flow-agent" -version = "2.0.3" +version = "2.0.5" source = { editable = "." } dependencies = [ { name = "cryptography" },