Port Omni Flash text-to-video to migrated Flow API

README.md

@@ -254,7 +254,8 @@ Three capabilities have no captured payload, so they fail with
 | 4K/1080p upscale (`/fk-pipeline` last step) | unported | none — keep the 1080p render |
 | 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 (`model_family=omni_flash`) | unported | use `model_family=veo` |
+| Omni Flash text-to-video | ported | `POST /api/flow/generate-video-omni-text` (4/6/8/10s) |
+| Omni Flash frame/reference modes | unported | use Veo or text-to-video until their batch payloads are captured |
 
 Restoring one starts with a capture, not a guess: [`docs/CAPTURE.md`](docs/CAPTURE.md).
 

agent/api/flow.py

@@ -9,6 +9,7 @@
     check_omni_flash_status,
     generate_omni_flash_first_frame_video,
     generate_omni_flash_first_last_video,
+    generate_omni_flash_text_video,
     generate_omni_flash_video,
 )
 
@@ -59,6 +60,15 @@ class GenerateOmniFlashVideoRequest(BaseModel):
     user_paygate_tier: str = "PAYGATE_TIER_ONE"
 
 
+class GenerateOmniFlashTextVideoRequest(BaseModel):
+    prompt: str
+    project_id: str
+    scene_id: str = ""
+    duration_s: int = 8
+    aspect_ratio: str = "VIDEO_ASPECT_RATIO_PORTRAIT"
+    user_paygate_tier: str = "PAYGATE_TIER_ONE"
+
+
 class UpscaleVideoRequest(BaseModel):
     media_id: str
     scene_id: str
@@ -219,6 +229,29 @@ async def generate_video_refs(body: GenerateVideoRefsRequest):
     return result.get("data", result)
 
 
+@router.post("/generate-video-omni-text")
+async def generate_video_omni_text(body: GenerateOmniFlashTextVideoRequest):
+    """Submit Omni 1.1 Flash text-to-video on flow.google.com.
+
+    Durations 4/6/8/10 seconds map to Flow's ``abra_t2v_<N>s`` models.
+    """
+    client = get_flow_client()
+    if not client.connected:
+        raise HTTPException(503, "Extension not connected")
+    try:
+        result = await generate_omni_flash_text_video(**body.model_dump())
+    except ValueError as exc:
+        raise HTTPException(400, str(exc)) from exc
+    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")),
+        )
+    return result.get("data", result)
+
+
 @router.post("/generate-video-omni")
 async def generate_video_omni(body: GenerateOmniFlashVideoRequest):
     """Submit Gemini Omni Flash reference-to-video generation.

agent/services/flow_batch.py

@@ -36,6 +36,7 @@
 
 RPC_GEN_IMAGE = "ogiZ0b"
 RPC_GEN_VIDEO = "eb1hJf"
+RPC_GEN_VIDEO_TEXT = "YhhmEf"
 RPC_OPERATION = "jwpduf"
 RPC_PROJECT_MEDIA = "Zzl0ze"
 RPC_MEDIA = "as29s"
@@ -357,6 +358,24 @@ def video_request(prompt: str, project_id: str, source_media_id: str,
     return build_envelope(RPC_GEN_VIDEO, inner)
 
 
+def text_video_request(prompt: str, project_id: str,
+                       aspect: Any = VIDEO_ASPECT_LANDSCAPE,
+                       model: str = "abra_t2v_4s") -> str:
+    """Build the migrated text-to-video submit (YhhmEf)."""
+    request = [
+        [None, None, [[[prompt]]]],
+        model,
+        resolve_video_aspect(aspect),
+        None,
+        [None, None, None, None, _client_uuid(), _client_uuid()],
+    ]
+    return build_envelope(RPC_GEN_VIDEO_TEXT, [
+        [request],
+        _context(project_id),
+        [_client_uuid(), 1],
+    ])
+
+
 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.
@@ -429,6 +448,26 @@ def read_uploaded_media_id(payload: Any) -> str:
     return media_id
 
 
+def read_text_video_submit(payload: Any) -> dict:
+    """Read YhhmEf's submitted media/workflow record."""
+    records = payload[3] if isinstance(payload, list) and len(payload) > 3 else None
+    record = records[0] if isinstance(records, list) and records else None
+    if not isinstance(record, list) or not record:
+        raise FlowBatchError("text-video submit carried no generation record")
+    media_id = record[0] if len(record) > 0 else None
+    project_id = record[1] if len(record) > 1 else None
+    workflow_id = record[2] if len(record) > 2 else None
+    status = record[3] if len(record) > 3 else None
+    if not isinstance(media_id, str) or not media_id:
+        raise FlowBatchError("text-video submit carried no media id")
+    return {
+        "media_id": media_id,
+        "project_id": project_id if isinstance(project_id, str) else None,
+        "workflow_id": workflow_id if isinstance(workflow_id, str) else media_id,
+        "status": status if isinstance(status, str) else None,
+    }
+
+
 def read_operation(payload: Any) -> Operation:
     """`[null, 50, [[opId, projectId, sceneId, status, …]]]`.
 

agent/services/omni_flash.py

@@ -27,6 +27,7 @@
 from urllib.parse import quote
 
 from agent.config import USE_BATCH_RPC
+from agent.services import flow_batch as fb
 from agent.services.flow_client import get_flow_client
 from agent.services.headers import random_headers
 
@@ -39,10 +40,10 @@
 #: captured off the new frontend, so on the batch path these fail with a
 #: name rather than dying on a 401 five retries deep.
 _UNSUPPORTED_ON_BATCH = (
-    "UNSUPPORTED_ON_BATCH_API: Omni Flash — it speaks the pre-migration REST "
-    "and tRPC endpoints, and no batchexecute payload for it has been captured; "
-    "see docs/CAPTURE.md. Use the Veo path (model_family=veo), or set "
-    "USE_BATCH_RPC=0 on a profile that still holds a bearer token."
+    "UNSUPPORTED_ON_BATCH_API: Omni Flash frame/reference generation is not yet "
+    "ported to flow.google.com batchexecute. Omni text-to-video is supported on "
+    "the batch path; frame-to-video, start+end and reference-to-video still need "
+    "their migrated payload captures."
 )
 
 
@@ -217,6 +218,54 @@ def _annotate_polling(result: dict, project_id: str) -> dict:
     return result
 
 
+async def generate_omni_flash_text_video(
+    prompt: str,
+    project_id: str,
+    scene_id: str = "",
+    duration_s: int = 8,
+    aspect_ratio: str = "VIDEO_ASPECT_RATIO_PORTRAIT",
+    user_paygate_tier: str = "PAYGATE_TIER_ONE",
+    seed: int | None = None,
+) -> dict:
+    """Submit Omni 1.1 Flash text-to-video on the migrated Flow batch API."""
+    _validate_duration(duration_s)
+    _validate_aspect(aspect_ratio)
+    if not USE_BATCH_RPC:
+        return {"error": "Omni text-to-video is implemented on the flow.google.com batch path only"}
+
+    client = get_flow_client()
+    try:
+        pid = client._batch_project_id(project_id)
+        model_key = f"abra_t2v_{duration_s}s"
+        freq = fb.text_video_request(prompt, pid, aspect=aspect_ratio, model=model_key)
+        payload = await client._batch_payload(
+            fb.RPC_GEN_VIDEO_TEXT, freq, fb.CAPTCHA_VIDEO, timeout=120)
+        submitted = fb.read_text_video_submit(payload)
+    except Exception as exc:
+        return {"status": 502, "error": f"{type(exc).__name__}: {exc}"}
+
+    media_id = submitted["media_id"]
+    workflow = {
+        "name": submitted.get("workflow_id") or media_id,
+        "primary_media_id": media_id,
+        "project_id": pid,
+    }
+    return {
+        "status": 200,
+        "data": {
+            "media": [{"name": media_id}],
+            "workflows": [workflow],
+            "model": model_key,
+            "duration_s": duration_s,
+            "flowkitPolling": {
+                "mode": "batch_media",
+                "project_id": pid,
+                "workflows": [workflow],
+            },
+        },
+    }
+
+
 async def _submit_omni_frame_video(
     *,
     start_image_media_id: str,
@@ -398,20 +447,68 @@ async def generate_omni_flash_video(
     return _annotate_polling(result, project_id)
 
 
-async def check_omni_flash_status(
+async def _check_omni_batch_media(
     workflows: list[dict],
     include_encoded_video: bool = False,
     project_id: str = "",
 ) -> dict:
-    """Perform one non-blocking poll pass for Omni workflow-backed jobs.
+    normalized = [item for workflow in (workflows or []) if (item := _normalize_workflow(workflow))]
+    if not normalized:
+        raise ValueError("Omni polling requires workflow descriptors with name and primary_media_id")
+    resolved_project_id = project_id or next(
+        (item.get("project_id", "") for item in normalized if item.get("project_id")), "")
+    client = get_flow_client()
+    items = []
+    for workflow in normalized:
+        media_id = workflow["primary_media_id"]
+        response = await client.get_media(media_id)
+        data = response.get("data") if isinstance(response, dict) else None
+        video = data.get("video") if isinstance(data, dict) else None
+        url = video.get("fifeUrl") if isinstance(video, dict) else None
+        if isinstance(url, str) and url.startswith("https://flow-content.google/video/"):
+            media = {
+                "media_id": media_id,
+                "url": url,
+                "encoded_video_available": False,
+                "resolved_via": "as29s",
+            }
+            if include_encoded_video:
+                media["encoded_video"] = None
+            items.append({
+                "name": workflow["name"],
+                "primary_media_id": media_id,
+                "project_id": workflow.get("project_id") or resolved_project_id,
+                "done": True,
+                "status": "MEDIA_GENERATION_STATUS_SUCCESSFUL",
+                "error": None,
+                "media": media,
+            })
+        else:
+            items.append({
+                "name": workflow["name"],
+                "primary_media_id": media_id,
+                "project_id": workflow.get("project_id") or resolved_project_id,
+                "done": False,
+                "status": "PENDING",
+                "error": None,
+            })
+    all_done = bool(items) and all(item["done"] for item in items)
+    return {
+        "project_id": resolved_project_id or None,
+        "done": all_done,
+        "status": "COMPLETED" if all_done else "PENDING",
+        "workflows": items,
+    }
 
-    Flow's production UI exposes workflow status through its authenticated
-    ``flow.projectInitialData`` tRPC response. The old ``/v1/media`` transport
-    currently returns ``INVALID_ARGUMENT`` for these workflow media IDs.
-    """
-    blocked = _batch_path_blocks_omni()
-    if blocked:
-        return blocked
+
+async def check_omni_flash_status(
+    workflows: list[dict],
+    include_encoded_video: bool = False,
+    project_id: str = "",
+) -> dict:
+    """Perform one non-blocking poll pass for Omni workflow-backed jobs."""
+    if USE_BATCH_RPC:
+        return await _check_omni_batch_media(workflows, include_encoded_video, project_id)
     normalized = []
     for workflow in workflows or []:
         item = _normalize_workflow(workflow)

docs/OMNI_FLASH.md

@@ -11,46 +11,55 @@ curl -fsS "$FLOWKIT_BASE_URL/health"
 curl -fsS "$FLOWKIT_BASE_URL/api/flow/status"
 ```
 
-Expected state:
+Expected state on the migrated Flow transport:
 
 ```json
 {"status":"ok","extension_connected":true}
-{"connected":true,"flow_key_present":true}
+{"connected":true,"transport":"batch"}
 ```
 
 Use `http://127.0.0.1:8100` when the caller runs on the FlowKit host. For a remote integration, set `FLOWKIT_BASE_URL` to the protected HTTPS reverse-proxy URL and allow only the required source IPs or private network. Do not expose Chrome, VNC/noVNC, the extension WebSocket, or port 8100 publicly.
 
 ## Supported modes
 
-| Mode | Inputs | Endpoint | Internal model family |
-|---|---|---|---|
-| First frame to video | one uploaded start image | `POST /api/flow/generate-video` | `abra_i2v_<duration>s` |
-| First + Last frame to video | uploaded start and end images | `POST /api/flow/generate-video` | `abra_i2v_<duration>s` |
-| References to video | 1-7 uploaded reference images | `POST /api/flow/generate-video-omni` | `abra_r2v_<duration>s` |
-
-Supported durations are `4`, `6`, `8`, and `10` seconds. Supported aspect ratios are:
+On `flow.google.com`, Omni **text-to-video** is migrated and live-verified. The older frame/reference implementations still use the pre-migration REST transport and remain fail-fast while `USE_BATCH_RPC=1`.
 
-- `VIDEO_ASPECT_RATIO_PORTRAIT` (`9:16`)
-- `VIDEO_ASPECT_RATIO_LANDSCAPE` (`16:9`)
+| Mode | Batch status | Endpoint | Internal model family |
+|---|---|---|---|
+| Text to video | **supported** | `POST /api/flow/generate-video-omni-text` | `abra_t2v_<duration>s` |
+| First frame to video | not yet ported | `POST /api/flow/generate-video` | `abra_i2v_<duration>s` (legacy only) |
+| First + Last frame to video | not yet ported | `POST /api/flow/generate-video` | `abra_i2v_<duration>s` (legacy only) |
+| References to video | not yet ported | `POST /api/flow/generate-video-omni` | `abra_r2v_<duration>s` (legacy only) |
 
-First + Last generation with `batchAsyncGenerateVideoStartAndEndImage` and the current `abra_i2v_*` mapping has been verified with a real Flow generation.
+Text-to-video supports `4`, `6`, `8`, and `10` seconds, with portrait and landscape aspect ratios. The migrated `YhhmEf` wire was live-verified with `abra_t2v_4s`; the downloaded result was exactly 4.000 seconds at 1280×720/24 fps. Completed media resolves through the migrated `as29s` media lookup.
 
 ## End-to-end integration flow
 
-An integration agent should implement this state machine:
-
 1. Check `/health` and `/api/flow/status`.
-2. Make each source image readable on the FlowKit server.
-3. Call `/api/flow/upload-image` for every source image and retain each returned `media_id`.
-4. Submit exactly one Omni request and persist its complete `flowkitPolling` object.
-5. Poll `/api/flow/check-omni-status` every 10-20 seconds using `project_id` and `workflows` from `flowkitPolling`.
-6. On `PENDING`, continue polling. On `FAILED`, stop and report the returned error. On `COMPLETED`, immediately download every non-null `media.url`.
-7. Store the downloaded video in the project's own durable storage. The returned Google URL is signed and short-lived.
+2. Submit `POST /api/flow/generate-video-omni-text` with prompt, project ID, duration and aspect ratio.
+3. Persist the returned `flowkitPolling` object.
+4. Poll `/api/flow/check-omni-status` every 10–20 seconds with its `project_id` and `workflows`.
+5. On `COMPLETED`, immediately download `workflows[].media.url`; the signed URL is short-lived.
 
-Do not send Omni workflow names to the legacy Veo `batchCheckAsyncVideoGenerationStatus` operation poller. Do not use the obsolete `/v1/media/<primaryMediaId>` polling path.
+Example:
+
+```bash
+curl -fsS -X POST "$FLOWKIT_BASE_URL/api/flow/generate-video-omni-text" \
+  -H 'Content-Type: application/json' \
+  -d '{
+    "prompt": "A small red paper boat gently drifts across a calm pond",
+    "project_id": "FLOW_PROJECT_ID",
+    "duration_s": 4,
+    "aspect_ratio": "VIDEO_ASPECT_RATIO_LANDSCAPE"
+  }'
+```
+
+Do not feed Omni workflow names to the legacy Veo operation poller.
 
 ## Supplying images
 
+This section applies to the legacy frame/reference Omni modes, which are not yet ported to the migrated batch transport.
+
 `POST /api/flow/upload-image` is not a multipart upload endpoint. Its `file_path` is an absolute path on the **FlowKit server**, not on the calling server.
 
 For a remote integration, first stage the file on the FlowKit host using an authenticated transfer such as SFTP/SCP, a private shared volume, or a separately secured upload service. Use a unique per-job directory, validate file size/type, and make the file readable by the FlowKit service account. Then call:

tests/unit/test_flow_batch.py

@@ -121,6 +121,22 @@ def test_a_hand_reframed_crop_overrides_the_default(self):
         crop = [None, 0.1, 1, 0.9]
         assert inner(fb.video_request("go", self.PID, "mid", crop=crop))[0][0][4][5] == crop
 
+    def test_text_video_matches_the_captured_yhhmef_shape(self):
+        payload = inner(fb.text_video_request(
+            "a boat", self.PID,
+            aspect="VIDEO_ASPECT_RATIO_LANDSCAPE",
+            model="abra_t2v_4s",
+        ))
+        request = payload[0][0]
+        assert request[0] == [None, None, [[["a boat"]]]]
+        assert request[1] == "abra_t2v_4s"
+        assert request[2] == fb.VIDEO_ASPECT_LANDSCAPE
+        assert request[3] is None
+        assert len(request[4]) == 6
+        assert payload[1][5] == self.PID
+        assert payload[2][1] == 1
+        assert fb.CAPTCHA_SLOT in json.dumps(payload)
+
 
 class TestReaders:
     OP = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
@@ -135,6 +151,15 @@ def test_a_repeated_url_is_not_a_second_variant(self):
         url = f"https://{fb.MEDIA_HOST}/image/{self.MID}?sig=x"
         assert len(fb.read_images([url, url])) == 1
 
+    def test_text_video_submit_reads_media_and_workflow_ids(self):
+        payload = [None, 10, [], [[self.MID, "project-1", self.OP, "CAE"]]]
+        assert fb.read_text_video_submit(payload) == {
+            "media_id": self.MID,
+            "project_id": "project-1",
+            "workflow_id": self.OP,
+            "status": "CAE",
+        }
+
     def test_operation_reads_the_id_and_status(self):
         op = fb.read_operation([None, 50, [[self.OP, "proj", "scene", "CAE"]]])
         assert (op.operation_id, op.status, op.done) == (self.OP, "CAE", True)