From 8566126358db59ffd4182d0fa37f3b2647866708 Mon Sep 17 00:00:00 2001 From: Bl0ck <36800583+Bl0ck154@users.noreply.github.com> Date: Tue, 8 Sep 2026 08:09:41 +0300 Subject: [PATCH] feat(flow): port Omni text video to migrated API --- README.md | 3 +- agent/api/flow.py | 33 +++++++++ agent/services/flow_batch.py | 39 +++++++++++ agent/services/omni_flash.py | 123 ++++++++++++++++++++++++++++++---- docs/OMNI_FLASH.md | 51 ++++++++------ tests/unit/test_flow_batch.py | 25 +++++++ tests/unit/test_omni_flash.py | 62 ++++++++++++++--- 7 files changed, 291 insertions(+), 45 deletions(-) diff --git a/README.md b/README.md index e4dd67ea..48c7a3b5 100644 --- a/README.md +++ b/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). diff --git a/agent/api/flow.py b/agent/api/flow.py index 99394096..c0a71025 100644 --- a/agent/api/flow.py +++ b/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_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. diff --git a/agent/services/flow_batch.py b/agent/services/flow_batch.py index f0111632..538f77e1 100644 --- a/agent/services/flow_batch.py +++ b/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, …]]]`. diff --git a/agent/services/omni_flash.py b/agent/services/omni_flash.py index 0db4c8cb..796d6a31 100644 --- a/agent/services/omni_flash.py +++ b/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) diff --git a/docs/OMNI_FLASH.md b/docs/OMNI_FLASH.md index 93aa1e94..64d3c040 100644 --- a/docs/OMNI_FLASH.md +++ b/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_s` | -| First + Last frame to video | uploaded start and end images | `POST /api/flow/generate-video` | `abra_i2v_s` | -| References to video | 1-7 uploaded reference images | `POST /api/flow/generate-video-omni` | `abra_r2v_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_s` | +| First frame to video | not yet ported | `POST /api/flow/generate-video` | `abra_i2v_s` (legacy only) | +| First + Last frame to video | not yet ported | `POST /api/flow/generate-video` | `abra_i2v_s` (legacy only) | +| References to video | not yet ported | `POST /api/flow/generate-video-omni` | `abra_r2v_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/` 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: diff --git a/tests/unit/test_flow_batch.py b/tests/unit/test_flow_batch.py index 63b8761d..a221023f 100644 --- a/tests/unit/test_flow_batch.py +++ b/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) diff --git a/tests/unit/test_omni_flash.py b/tests/unit/test_omni_flash.py index 2333e6ee..cf3b2fb1 100644 --- a/tests/unit/test_omni_flash.py +++ b/tests/unit/test_omni_flash.py @@ -20,6 +20,7 @@ extract_omni_workflows, generate_omni_flash_first_frame_video, generate_omni_flash_first_last_video, + generate_omni_flash_text_video, generate_omni_flash_video, ) @@ -132,6 +133,36 @@ def _mock_submit_client(): return client +@pytest.mark.asyncio +async def test_batch_text_video_builds_4s_yhhmef_submit(monkeypatch): + monkeypatch.setattr(omni_flash, "USE_BATCH_RPC", True) + client = MagicMock() + client._batch_project_id.return_value = "11111111-2222-3333-4444-555555555555" + client._batch_payload = AsyncMock(return_value=[ + None, 10, [], [[ + "22222222-3333-4444-5555-666666666666", + "11111111-2222-3333-4444-555555555555", + "77777777-8888-9999-aaaa-bbbbbbbbbbbb", "CAE", + ]], + ]) + with patch("agent.services.omni_flash.get_flow_client", return_value=client): + result = await generate_omni_flash_text_video( + prompt="A red paper boat drifts across a pond", + project_id="11111111-2222-3333-4444-555555555555", + duration_s=4, + aspect_ratio="VIDEO_ASPECT_RATIO_LANDSCAPE", + ) + assert result["status"] == 200 + assert result["data"]["model"] == "abra_t2v_4s" + assert result["data"]["duration_s"] == 4 + assert result["data"]["flowkitPolling"]["mode"] == "batch_media" + rpcid, freq, captcha = client._batch_payload.await_args.args[:3] + assert rpcid == omni_flash.fb.RPC_GEN_VIDEO_TEXT + assert captcha == omni_flash.fb.CAPTCHA_VIDEO + payload = __import__("json").loads(__import__("json").loads(freq)[0][0][1]) + assert payload[0][0][1] == "abra_t2v_4s" + + @pytest.mark.asyncio async def test_submit_builds_flow_omni_first_frame_request_and_poll_descriptor(): client = _mock_submit_client() @@ -353,6 +384,23 @@ async def test_media_redirect_fetch_requests_url_only_mode(): ) +@pytest.mark.asyncio +async def test_batch_omni_poll_uses_as29s_media(monkeypatch): + monkeypatch.setattr(omni_flash, "USE_BATCH_RPC", True) + client = MagicMock() + client.get_media = AsyncMock(return_value={ + "status": 200, + "data": {"video": {"fifeUrl": "https://flow-content.google/video/media-1?Signature=test"}}, + }) + with patch("agent.services.omni_flash.get_flow_client", return_value=client): + result = await check_omni_flash_status([{ + "name": "workflow-1", "primary_media_id": "media-1", "project_id": "project-1", + }]) + assert result["done"] is True + assert result["workflows"][0]["media"]["resolved_via"] == "as29s" + client.get_media.assert_awaited_once_with("media-1") + + @pytest.mark.asyncio async def test_omni_poll_pending_uses_project_snapshot_not_legacy_transports(): client = MagicMock() @@ -477,7 +525,7 @@ async def test_submit_rejects_empty_reference_set(): ) -class TestBatchPathIsRefusedRatherThanAttempted: +class TestUnportedOmniBatchModesAreRefusedRatherThanAttempted: """Flow stopped minting the bearer these endpoints need, and no Omni payload has been captured off the new frontend. Saying so beats a 401 five retries deep.""" @@ -513,14 +561,8 @@ async def test_reference_to_video_names_the_gap_and_sends_nothing(self, client): assert "UNSUPPORTED_ON_BATCH_API" in result["error"] client._send.assert_not_called() - async def test_polling_names_the_gap_and_sends_nothing(self, client): - result = await check_omni_flash_status( - [{"name": "wf", "primary_media_id": "mid", "project_id": "pid"}]) - assert "UNSUPPORTED_ON_BATCH_API" in result["error"] - client._send.assert_not_called() - - async def test_the_message_points_at_both_ways_out(self, client): + async def test_the_message_points_to_supported_text_to_video(self, client): result = await generate_omni_flash_video( reference_media_ids=["a"], prompt="go", project_id="pid") - assert "docs/CAPTURE.md" in result["error"] - assert "USE_BATCH_RPC=0" in result["error"] + assert "text-to-video is supported" in result["error"] + assert "reference-to-video" in result["error"]