Restore native 1080p export on migrated Flow API

README.md

@@ -254,12 +254,13 @@ without being silently replaced by the default model. See
 
 ### What does not work on the new API yet
 
-Three capabilities have no captured payload, so they fail with
-`UNSUPPORTED_ON_BATCH_API` rather than quietly producing the wrong thing:
+1080p export is ported: FlowKit mirrors the Flow UI's `p0UkFb` high-resolution
+Download request and polls the resulting media through `as29s`. The remaining
+capabilities below still fail loudly rather than quietly producing the wrong thing:
 
 | Capability | Status | Workaround |
 |---|---|---|
-| 4K/1080p upscale (`/fk-pipeline` last step) | unported | none — keep the 1080p render |
+| 4K export | plan-gated / not live-verified | use 1080p; Google Flow exposes Full HD as the standard high-resolution export |
 | Reference-to-video (r2v) | unported | `FLOW_ALLOW_DEGRADED=1` → i2v off the first reference |
 | Start+end-frame chaining (`/fk-gen-chain-videos`) | unported | `FLOW_ALLOW_DEGRADED=1` → i2v off the start frame |
 | Omni Flash text-to-video | ported | `POST /api/flow/generate-video-omni-text` (4/6/8/10s) |

agent/api/flow.py

@@ -5,6 +5,7 @@
 
 from agent.config import USE_BATCH_RPC, FLOW_PROJECT_ID, FLOW_ALLOW_DEGRADED
 from agent.services.flow_client import get_flow_client
+from agent.services.flow_poll import annotate_polling, check_workflow_status
 from agent.services.omni_flash import (
     check_omni_flash_status,
     generate_omni_flash_first_frame_video,
@@ -78,6 +79,15 @@ class UpscaleVideoRequest(BaseModel):
     scene_id: str
     aspect_ratio: str = "VIDEO_ASPECT_RATIO_PORTRAIT"
     resolution: str = "VIDEO_RESOLUTION_4K"
+    project_id: Optional[str] = None
+
+
+class ExportVideoRequest(BaseModel):
+    media_id: str
+    scene_id: str = "export"
+    quality: Literal["1080p", "4k"] = "1080p"
+    aspect_ratio: str = "VIDEO_ASPECT_RATIO_LANDSCAPE"
+    project_id: Optional[str] = None
 
 
 class UploadImageRequest(BaseModel):
@@ -94,6 +104,7 @@ class CheckStatusRequest(BaseModel):
     workflows: Optional[list[dict]] = None
     project_id: str = ""
     include_encoded_video: bool = False
+    mode: Literal["omni", "export"] = "omni"
 
 
 class CheckOmniStatusRequest(BaseModel):
@@ -300,19 +311,61 @@ async def upscale_video(body: UpscaleVideoRequest):
     return result.get("data", result)
 
 
+@router.post("/export-video")
+async def export_video(body: ExportVideoRequest):
+    """Start Google's native Full HD/4K export and return workflow polling data."""
+    client = get_flow_client()
+    if not client.connected:
+        raise HTTPException(503, "Extension not connected")
+    resolution = "VIDEO_RESOLUTION_1080P" if body.quality == "1080p" else "VIDEO_RESOLUTION_4K"
+    result = await client.upscale_video(
+        media_id=body.media_id,
+        scene_id=body.scene_id,
+        aspect_ratio=body.aspect_ratio,
+        resolution=resolution,
+        project_id=body.project_id,
+    )
+    if result.get("error") or (
+        isinstance(result.get("status"), int) and result["status"] >= 400
+    ):
+        raise HTTPException(result.get("status", 502), result.get("error", result.get("data")))
+
+    annotated = annotate_polling(result, mode="media_redirect")
+    data = annotated.get("data", annotated)
+    if isinstance(data, dict):
+        data["export"] = {
+            "quality": body.quality,
+            "resolution": resolution,
+            "native_flow_export": True,
+            "next": "/api/flow/check-status",
+            "check_status_mode": "export",
+        }
+    return data
+
+
 @router.post("/check-status")
 async def check_status(body: CheckStatusRequest):
-    """Check Veo operation status or Omni workflow/media status.
+    """Check Veo operations, Omni workflows, or native export workflows.
 
-    Veo: pass ``operations``.
-    Omni Flash: pass ``workflows`` from submit ``flowkitPolling.workflows``.
+    Veo: pass ``operations``. Workflow callers pass ``workflows``; native video
+    export additionally sets ``mode=\"export\"``.
     """
     client = get_flow_client()
     if not client.connected:
         raise HTTPException(503, "Extension not connected")
 
     if body.workflows:
         try:
+            if body.mode == "export":
+                result = await check_workflow_status(
+                    body.workflows,
+                    mode="media_redirect",
+                    include_encoded_video=body.include_encoded_video,
+                    project_id=body.project_id,
+                    client=client,
+                )
+                result["download_ready"] = result.get("status") == "COMPLETED"
+                return result
             return await check_omni_flash_status(
                 body.workflows,
                 include_encoded_video=body.include_encoded_video,

agent/services/flow_batch.py

@@ -42,6 +42,7 @@
 RPC_MEDIA = "as29s"
 RPC_UPLOAD_IMAGE = "maseQ"
 RPC_UPSCALE_IMAGE = "SPrCad"
+RPC_UPSCALE = "p0UkFb"
 
 CAPTCHA_IMAGE = "IMAGE_GENERATION"
 CAPTCHA_VIDEO = "VIDEO_GENERATION"
@@ -429,6 +430,23 @@ def text_video_request(prompt: str, project_id: str,
     ])
 
 
+def upscale_request(media_id: str, project_id: str,
+                    aspect: Any = VIDEO_ASPECT_LANDSCAPE,
+                    model: str = "veo_3_1_upsampler_1080p") -> str:
+    """Build Flow's migrated high-resolution download request (RPC p0UkFb)."""
+    item = [None] * 32
+    item[0] = [None, media_id]
+    item[2] = 1
+    item[4] = [None, str(uuid.uuid4()), None, None, _client_uuid()]
+    item[6] = resolve_video_aspect(aspect)
+    item[31] = model
+    return build_envelope(RPC_UPSCALE, [
+        [item],
+        _context(project_id),
+        [_client_uuid()],
+    ])
+
+
 def upload_request(image_b64: str, project_id: str, mime_type: str = "image/jpeg",
                    file_name: str = "upload.jpg") -> str:
     """Put a local image into the project so it can be used as a reference.
@@ -528,6 +546,14 @@ def read_upscaled_image(payload: Any) -> str:
     return encoded
 
 
+def read_upscaled_media_id(payload: Any) -> str:
+    """Return the media id created by the p0UkFb upscale submit."""
+    for text in _walk_strings(payload):
+        if isinstance(text, str) and text.endswith("_upsampled"):
+            return text
+    raise FlowBatchError("upscale response carried no upsampled media id")
+
+
 def read_operation(payload: Any) -> Operation:
     """`[null, 50, [[opId, projectId, sceneId, status, …]]]`.
 

agent/services/flow_client.py

@@ -20,6 +20,7 @@
 import time
 import uuid
 from typing import Optional
+from urllib.parse import quote
 
 from agent.config import (
     GOOGLE_FLOW_API, GOOGLE_API_KEY, ENDPOINTS,
@@ -839,14 +840,33 @@ async def generate_video_from_references(self, reference_media_ids: list[str],
 
     async def upscale_video(self, media_id: str, scene_id: str,
                              aspect_ratio: str = "VIDEO_ASPECT_RATIO_PORTRAIT",
-                             resolution: str = "VIDEO_RESOLUTION_4K") -> dict:
-        """Upscale a video."""
+                             resolution: str = "VIDEO_RESOLUTION_4K",
+                             project_id: str | None = None) -> dict:
+        """Upscale/export a video using Flow's migrated p0UkFb RPC."""
         if not USE_BATCH_RPC:
             return await self._legacy_upscale_video(media_id, scene_id, aspect_ratio, resolution)
-        return {"error": _unsupported(
-            "video upscale",
-            "no upsampler rpc appears in the new frontend's captures",
-        )}
+
+        model = UPSCALE_MODELS.get(resolution)
+        if not model:
+            return {"status": 400, "error": f"Unsupported upscale resolution: {resolution}"}
+        try:
+            pid = self._batch_project_id(project_id or "")
+            freq = fb.upscale_request(media_id, pid, aspect=aspect_ratio, model=model)
+            payload = await self._batch_payload(
+                fb.RPC_UPSCALE, freq, fb.CAPTCHA_VIDEO, timeout=120)
+            upscaled_media_id = fb.read_upscaled_media_id(payload)
+        except Exception as e:
+            return _batch_error(e)
+
+        workflow = {
+            "name": upscaled_media_id,
+            "primary_media_id": upscaled_media_id,
+            "project_id": pid,
+        }
+        return {"status": 200, "data": {
+            "media": [{"name": upscaled_media_id}],
+            "workflows": [workflow],
+        }}
 
     async def check_video_status(self, operations: list[dict]) -> dict:
         """One poll round for each submitted operation.
@@ -1016,6 +1036,54 @@ async def get_media(self, media_id: str) -> dict:
             data["image"] = {"fifeUrl": urls.image}
         return {"status": 200, "data": data}
 
+    async def resolve_media_url(self, media_id: str) -> dict:
+        """Resolve a signed Flow media URL without downloading the media body."""
+        if USE_BATCH_RPC:
+            result = await self.get_media(media_id)
+            data = result.get("data") if isinstance(result.get("data"), dict) else {}
+            video = data.get("video") if isinstance(data, dict) else None
+            candidate = video.get("fifeUrl") if isinstance(video, dict) else None
+            return {
+                "status": result.get("status", 200),
+                "data": {
+                    "url": candidate,
+                    "contentType": "video/mp4" if candidate else None,
+                },
+                "error": result.get("error"),
+            }
+
+        url = (
+            "https://labs.google/fx/api/trpc/media.getMediaUrlRedirect"
+            f"?name={quote(media_id, safe='')}"
+        )
+        return await self._send(
+            "trpc_request",
+            {
+                "url": url,
+                "method": "GET",
+                "headers": {"content-type": "application/json"},
+                "responseMode": "url",
+            },
+            timeout=15,
+        )
+
+    async def get_project_initial_data(self, project_id: str) -> dict:
+        """Fetch the authenticated legacy Flow project snapshot used for polling."""
+        query = quote(
+            json.dumps({"json": {"projectId": project_id}}, separators=(",", ":")),
+            safe="",
+        )
+        url = f"https://labs.google/fx/api/trpc/flow.projectInitialData?input={query}"
+        return await self._send(
+            "trpc_request",
+            {
+                "url": url,
+                "method": "GET",
+                "headers": {"content-type": "application/json"},
+            },
+            timeout=15,
+        )
+
     async def upload_image(self, image_base64: str, mime_type: str = "image/jpeg",
                             project_id: str = "", file_name: str = "image.jpg") -> dict:
         """Upload an image into the project so it can be used as a reference."""