diff --git a/.gitignore b/.gitignore index 766c04bdc..636c99a04 100644 --- a/.gitignore +++ b/.gitignore @@ -21,4 +21,6 @@ GEMINI.md ds-bundle/ .design-sync/.cache/ .design-sync/learnings/ -.design-sync/node_modules \ No newline at end of file +.design-sync/node_modules +# Chrome generates this when it loads the unpacked extension +extension/_metadata/ diff --git a/README.md b/README.md index 8084b1de3..04be57229 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,23 @@ --- +> **Working against the new Google Flow API.** Flow moved to `flow.google.com` +> in September 2026 and stopped minting the `Bearer ya29.…` that the old +> `aisandbox-pa.googleapis.com` REST API needed. The `batchexecute` transport +> that replaces it is in place and verified end to end against the live API — +> image generation, 2K image export, and image-to-video all run green. Upgrading +> from an older Flow Kit: reload the extension (v0.3.2+) and pin +> `FLOW_PROJECT_ID`; Flow Kit can no longer create the project for you. +> +> Four capabilities are still unported because their payloads were never +> captured off the new UI — **video upscale**, **reference-to-video (r2v)**, +> **start+end-frame chaining**, and **Omni Flash frame/reference**. They fail +> loudly with `UNSUPPORTED_ON_BATCH_API` instead of quietly producing the wrong +> thing. `FLOW_ALLOW_DEGRADED=1` drops chaining and r2v to plain i2v; video +> upscale has no fallback. To restore one properly see [`docs/CAPTURE.md`](docs/CAPTURE.md). + +--- + ### ☕ Sponsor this project @@ -188,9 +205,9 @@ One signed-in Flow tab has to stay open; nothing here works headless. > **September 2026 — Flow moved.** It now lives at `flow.google.com` and the old > `aisandbox-pa.googleapis.com` REST API has no caller: the `Bearer ya29.…` it > needed stopped being minted. If you are upgrading from an older Flow Kit, -> reload the extension (v0.3.0+) and pin `FLOW_PROJECT_ID` — see -> [Configuration](#configuration). The legacy path is still there behind -> `USE_BATCH_RPC=0`, but it is a post-mortem tool, not a fallback. +> reload the extension (v0.3.2+) and pin `FLOW_PROJECT_ID` — see +> [Configuration](#configuration). The REST path has been removed; `git log` +> has it if a payload is ever needed for reference. ## Quick Start @@ -240,7 +257,6 @@ You can also pass `flow_project_id` per project on `POST /api/projects`. | Env var | Default | What it does | |---------|---------|--------------| | `FLOW_PROJECT_ID` | — | The Flow project every RPC is scoped to. Required. | -| `USE_BATCH_RPC` | `1` | `0` falls back to the pre-migration REST path (dead auth). | | `FLOW_ALLOW_DEGRADED` | `0` | `1` lets scene chaining and r2v fall back to plain i2v instead of failing. | | `DEFAULT_PAYGATE_TIER` | `PAYGATE_TIER_TWO` | Carried for the DB and dashboard; no longer selects a model. | @@ -690,7 +706,6 @@ agent/ │ # flow, models, providers, reviews, materials, music, tts) ├── services/ │ ├── flow_client.py # WS bridge to extension -│ ├── headers.py # Randomized browser headers │ ├── tts.py # OmniVoice TTS (subprocess-based) │ ├── scene_chain.py # Continuation scene logic │ ├── video_reviewer.py # AI vision review — contact sheet + claude/agy/codex CLI dispatch @@ -805,13 +820,13 @@ These arrive in the response body as `data.error.details[].reason`. The worker a | Status | Source | Meaning | Handling | |--------|--------|---------|----------| | **400** | Flow API | Invalid payload, UNSAFE_GENERATION, entity not found (sometimes) | Route by `details.reason` — some are auto-recoverable, others terminal | -| **401** | Flow API (legacy path only) | Bearer expired — on a post-migration profile it was never minted | Switch to the batch path (`USE_BATCH_RPC=1`) | +| **401** | Flow API | Should not occur — batchexecute authenticates in the page, not with a bearer | Check the Flow tab is signed in; see `NO_AT_TOKEN` | | **403** | Extension (`background.js:432`) | `CAPTCHA_FAILED`, `NO_FLOW_TAB`, or `MODEL_ACCESS_DENIED` | CAPTCHA → retry loop; NO_FLOW_TAB → fail (user must open Flow); tier → fail | | **404** | Flow API | `media_id` not found (expired upload) | Same as "Requested entity was not found" — auto re-upload | | **429** | Flow API | Rate limited / quota | Back off + retry; if `USER_QUOTA_REACHED` appears, fail | | **500** | Flow backend **or** extension fetch exception (`background.js:504`) | Transient server error OR network drop during fetch | Retry with exponential backoff | | **502** | FastAPI default (`agent/api/flow.py:80,92`) | Extension returned error without explicit status | Retry; check extension health | -| **503** | FastAPI (`api/flow.py`) | "Extension not connected" or `NO_FLOW_KEY` | Worker waits for reconnect — status set to PENDING, not FAILED | +| **503** | FastAPI (`api/flow.py`) | "Extension not connected" | Worker waits for reconnect — status set to PENDING, not FAILED | | **504** | Agent | 60s timeout waiting for extension WS response | Treated as transient; re-queue PENDING | Status-code detection logic lives in `agent/worker/_parsing.py:_is_error` — a result is an error if `result.error` is set, `status >= 400`, **or** `data.error` is present. @@ -825,7 +840,6 @@ String patterns in `error_message` that the worker recognizes: | `Extension not connected` | Chrome extension offline or WS dropped | 503 returned; worker re-queues PENDING and waits | | `extension reconnected` / `extension disconnected` | WS bounce mid-request | Re-queue PENDING without incrementing `retry_count` | | `extension_switched` | User switched Flow tabs mid-generation | Re-queue PENDING | -| `NO_FLOW_KEY` | Extension has no captured bearer token — **legacy path only**, expected on the batch path | Only meaningful with `USE_BATCH_RPC=0` | | `NO_AT_TOKEN` | Flow tab is signed out, on an interstitial, or still booting | Open `flow.google.com`, sign in, let the app load | | `NO_FLOW_PROJECT` | No Flow project to scope the RPC to | Pin `FLOW_PROJECT_ID` — **terminal, not retried** | | `UNSUPPORTED_ON_BATCH_API` | Upscale / r2v / chaining — payload never captured | See `docs/CAPTURE.md` — **terminal, not retried** | diff --git a/agent/api/flow.py b/agent/api/flow.py index 5aeed8fac..0d94b780d 100644 --- a/agent/api/flow.py +++ b/agent/api/flow.py @@ -3,7 +3,7 @@ from pydantic import BaseModel, Field from typing import Literal, Optional -from agent.config import USE_BATCH_RPC, FLOW_PROJECT_ID, FLOW_ALLOW_DEGRADED +from agent.config import FLOW_PROJECT_ID, FLOW_ALLOW_DEGRADED from agent.services.flow_client import get_flow_client from agent.services.omni_flash import ( check_omni_flash_status, @@ -122,15 +122,17 @@ class UpscaleImageRequest(BaseModel): @router.get("/status") async def extension_status(): - """Extension health, and which transport it is being asked to speak. + """Extension health. - `flow_key_present` is a legacy-path signal: the batchexecute path has no - bearer token at all, so false is expected there rather than a fault. + `flow_key_present: false` is expected, not a fault: batchexecute + authenticates in the page and there is no bearer token to capture. """ client = get_flow_client() return { "connected": client.connected, - "transport": "batch" if USE_BATCH_RPC else "legacy_rest", + # One transport now. The key stays so the documented pre-flight check + # (CLAUDE.md) keeps reading {"transport": "batch", ...}. + "transport": "batch", "flow_project_id": FLOW_PROJECT_ID or None, "allow_degraded": FLOW_ALLOW_DEGRADED, "flow_key_present": client._flow_key is not None, diff --git a/agent/api/projects.py b/agent/api/projects.py index 600ebe8a9..4b87aa847 100644 --- a/agent/api/projects.py +++ b/agent/api/projects.py @@ -7,7 +7,7 @@ from fastapi import APIRouter, HTTPException from pydantic import BaseModel -from agent.config import BASE_DIR, USE_BATCH_RPC +from agent.config import BASE_DIR from agent.models.project import Project, ProjectCreate, ProjectUpdate from agent.models.character import Character from agent.sdk.persistence.sqlite_repository import SQLiteRepository @@ -172,9 +172,9 @@ async def create(body: ProjectCreate): detected_tier = await _detect_user_tier(client) - # On the batch path Flow no longer creates projects for us — the uuid comes - # from the request or from FLOW_PROJECT_ID. The legacy path still mints one. - flow_project_id = client.flow_project_id(body.flow_project_id) if USE_BATCH_RPC else None + # Flow no longer creates projects for us — the uuid comes from the request + # or from FLOW_PROJECT_ID. + flow_project_id = client.flow_project_id(body.flow_project_id) if flow_project_id: logger.info("Flow project reused: %s", flow_project_id) else: diff --git a/agent/config.py b/agent/config.py index a02a51bd7..743198e2d 100644 --- a/agent/config.py +++ b/agent/config.py @@ -15,19 +15,12 @@ WS_HOST = os.environ.get("WS_HOST", "127.0.0.1") WS_PORT = int(os.environ.get("WS_PORT", "9222")) -# ─── Google Flow API ──────────────────────────────────────── -# Legacy REST host. Flow moved to flow.google.com in September 2026 and stopped -# minting the `Bearer ya29.…` this host needs, so these are only reachable with -# USE_BATCH_RPC=0 on a browser profile that still has an old token. -GOOGLE_FLOW_API = "https://aisandbox-pa.googleapis.com" -GOOGLE_API_KEY = os.environ.get("GOOGLE_API_KEY", "AIzaSyBtrm0o5ab1c-Ec8ZuLcGt3oJAA5VWt3pY") -RECAPTCHA_SITE_KEY = os.environ.get("RECAPTCHA_SITE_KEY", "6LdsFiUsAAAAAIjVDZcuLhaHiDn5nnHVXVRQGeMV") - -# ─── Flow batchexecute (the current path) ─────────────────── + +# ─── Flow batchexecute ────────────────────────────────────── # Every call is signed in the page with the session cookie plus a per-page `at` -# token, so the extension runs it inside a signed-in flow.google.com tab. Set -# USE_BATCH_RPC=0 only to fall back to the dead REST path for a post-mortem. -USE_BATCH_RPC = os.environ.get("USE_BATCH_RPC", "1") == "1" +# token, so the extension runs it inside a signed-in flow.google.com tab. This +# is the only transport; the REST path it replaced was removed once Flow stopped +# minting the bearer it needed. # The Flow project every RPC is scoped to. Project creation went with the old # labs.google tRPC endpoint, so a project is made once in the Flow UI and its @@ -66,20 +59,6 @@ # newly introduced image models do not require a Flow Kit release. DEFAULT_IMAGE_MODEL = _MODELS.get("default_image_model", "NANO_BANANA_PRO") -# ─── API Endpoints ─────────────────────────────────────────── -ENDPOINTS = { - "generate_images": "/v1/projects/{project_id}/flowMedia:batchGenerateImages", - "generate_video": "/v1/video:batchAsyncGenerateVideoStartImage", - "generate_video_start_end": "/v1/video:batchAsyncGenerateVideoStartAndEndImage", - "generate_video_references": "/v1/video:batchAsyncGenerateVideoReferenceImages", - "upscale_video": "/v1/video:batchAsyncGenerateVideoUpsampleVideo", - "upscale_image": "/v1/flow/upsampleImage", - "upload_image": "/v1/flow/uploadImage", - "check_video_status": "/v1/video:batchCheckAsyncVideoGenerationStatus", - "get_credits": "/v1/credits", - "get_media": "/v1/media/{media_id}", -} - # ─── Output Directories ───────────────────────────────────── OUTPUT_DIR = BASE_DIR / "output" SHARED_OUTPUT_DIR = OUTPUT_DIR / "_shared" @@ -131,29 +110,3 @@ def _load_suno_key() -> str: SUNO_POLL_INTERVAL = int(os.environ.get("SUNO_POLL_INTERVAL", "5")) SUNO_POLL_TIMEOUT = int(os.environ.get("SUNO_POLL_TIMEOUT", "600")) -# ─── Header Randomization Pools ───────────────────────────── -USER_AGENTS = [ - "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36", - "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36", - "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36", - "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36", - "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36", - "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36", -] - -CHROME_VERSIONS = [ - '"Google Chrome";v="109", "Chromium";v="109"', - '"Google Chrome";v="110", "Chromium";v="110"', - '"Google Chrome";v="111", "Chromium";v="111"', - '"Google Chrome";v="113", "Not-A.Brand";v="24"', - '"Google Chrome";v="120", "Not-A.Brand";v="24"', - '"Google Chrome";v="141", "Not?A_Brand";v="8", "Chromium";v="141"', -] - -BROWSER_VALIDATIONS = [ - "SgDQo8mvrGRdD61Pwo8wyWVgYgs=", -] - -CLIENT_DATA = [ - "CKi1yQEIh7bJAQiktskBCKmdygEIvorLAQiUocsBCIagzQEYv6nKARjRp88BGKqwzwE=", -] diff --git a/agent/sdk/services/operations.py b/agent/sdk/services/operations.py index b0199e82e..e7157cd45 100644 --- a/agent/sdk/services/operations.py +++ b/agent/sdk/services/operations.py @@ -40,7 +40,7 @@ def _char_matches(c: dict, name_set: set) -> bool: import aiohttp from agent.db import crud -from agent.config import USE_BATCH_RPC, VIDEO_POLL_INTERVAL, VIDEO_POLL_TIMEOUT +from agent.config import VIDEO_POLL_INTERVAL, VIDEO_POLL_TIMEOUT from agent.utils.paths import scene_4k_path from agent.utils.slugify import slugify from agent.worker._parsing import ( @@ -447,28 +447,19 @@ async def generate_scene_video(self, scene: dict, orientation: str, base_prompt = scene.get("video_prompt") or scene.get("prompt", "") prompt = await _build_video_prompt(base_prompt, scene, pid) - # Check if already submitted (op_name saved from previous attempt) - # OLD schema (Lite/Fast/Ultra): op_name is "models/.../operations/..." → re-poll via check_video_status - # NEW schema (Low Priority workflow): op_name is bare UUID → cannot recover (need primary_media_id - # which isn't persisted yet); fall through and resubmit (Low Priority is free, duplicate is OK) + # Already submitted on a previous attempt? Re-poll it. existing_op = None if request_id: req_row = await crud.get_request(request_id) existing_op = req_row.get("request_id") if req_row else None - # A bare uuid means different things on the two transports. On the - # legacy path it is a Low Priority workflow name that cannot be - # re-polled, so the retry resubmits. On the batch path it is the - # operation id, and looking it up in the project listing is exactly - # what the status poll does — resubmitting there would abandon a - # running render and pay for a second one. - bare_uuid = bool(existing_op and len(existing_op) == 36 and existing_op.count("-") == 4) - looks_like_workflow_uuid = bare_uuid and not USE_BATCH_RPC - if existing_op and not looks_like_workflow_uuid: + # A bare uuid is the operation id, and looking it up in the project + # listing is exactly what the status poll does — resubmitting instead + # would abandon a running render and pay for a second one. + if existing_op: logger.info("Video gen already submitted (op=%s), re-polling", existing_op[:30]) operations = [{"operation": {"name": existing_op}, "status": "MEDIA_GENERATION_STATUS_PENDING"}] return await _poll_operations(self._client, operations) - # else: workflow UUID — fall through and resubmit fresh submit_result = await self._client.generate_video( start_image_media_id=image_media_id, diff --git a/agent/services/flow_client.py b/agent/services/flow_client.py index be56e708a..92c1b003f 100644 --- a/agent/services/flow_client.py +++ b/agent/services/flow_client.py @@ -4,15 +4,18 @@ Agent runs a WS server. Extension connects as client. Agent sends requests, extension executes them in browser context (residential IP, cookies, reCAPTCHA). -Two transports live here. The current one is Flow's ``batchexecute`` endpoint on -flow.google.com, whose calls only a signed-in page can sign — the agent builds -the envelope, the extension runs it in the tab (see :mod:`agent.services.flow_batch`). -The old REST path against ``aisandbox-pa.googleapis.com`` is kept behind -``USE_BATCH_RPC=0``; it needs a ``Bearer ya29.…`` that Flow stopped minting in -the September 2026 migration, so it is a post-mortem tool, not a fallback. - -Both shape their answers the same way, so everything downstream — the worker's -parsers, the operation poller, the scene/character updaters — is transport-blind. +One transport: Flow's ``batchexecute`` endpoint on flow.google.com, whose calls +only a signed-in page can sign — the agent builds the envelope, the extension +runs it in the tab (see :mod:`agent.services.flow_batch`). + +The REST path against ``aisandbox-pa.googleapis.com`` that preceded it is gone. +It needed a ``Bearer ya29.…`` that Flow stopped minting in the September 2026 +migration, so it could not run; keeping it only gave the next contributor a +second place to implement things. Git history has it if a payload is ever needed. + +Answers are shaped like the old REST ones, so everything downstream — the +worker's parsers, the operation poller, the scene/character updaters — reads one +shape and never learns where it came from. """ import asyncio import json @@ -22,14 +25,12 @@ from typing import Optional from agent.config import ( - GOOGLE_FLOW_API, GOOGLE_API_KEY, ENDPOINTS, - VIDEO_MODELS, UPSCALE_MODELS, IMAGE_MODELS, VIDEO_POLL_TIMEOUT, - USE_BATCH_RPC, FLOW_PROJECT_ID, FLOW_ALLOW_DEGRADED, + VIDEO_MODELS, + FLOW_PROJECT_ID, FLOW_ALLOW_DEGRADED, DEFAULT_PAYGATE_TIER, ) from agent import config as _config from agent.services import flow_batch as fb -from agent.services.headers import random_headers logger = logging.getLogger(__name__) @@ -368,17 +369,8 @@ async def refresh_project_urls(self, project_id: str) -> dict: The batch path can do this properly: the media rpc answers a media id with a freshly signed url, so we walk the project's scenes and entities - and refresh each id we hold. The legacy path could not — its media - endpoint returned base64 content rather than a url — so it still asks - the user to open the project in Chrome and let the intercept catch them. + and refresh each id we hold. """ - if not USE_BATCH_RPC: - logger.info("URL refresh requested for project %s — legacy path has no " - "url-serving media endpoint", project_id[:12]) - return {"refreshed": 0, "found": 0, "note": "Legacy REST path: no URL refresh. " - "Open the project in Google Flow in Chrome and let the extension " - "intercept fresh URLs, or set USE_BATCH_RPC=1."} - from agent.db import crud # (media_id, kind) -> the scene/character fields it should land in @@ -434,14 +426,10 @@ async def _send(self, method: str, params: dict, timeout: float = 300) -> dict: if not self.connected: return {"error": "Extension not connected"} - # A bearer token is only worth routing on when something is going to - # send one. The batchexecute path authenticates in the page with the - # session cookie, so demanding a flow key there would reject every - # profile — no profile on that path ever captures one. - needs_token = not USE_BATCH_RPC - extension_candidates = self._extension_candidates(require_token=needs_token) - if not extension_candidates and needs_token: - return {"error": "NO_FLOW_KEY"} + # No profile needs a bearer any more: batchexecute authenticates in the + # page with the session cookie. Demanding a flow key here would reject + # every profile, because none on this path ever captures one. + extension_candidates = self._extension_candidates(require_token=False) if not extension_candidates: return {"error": "Extension not connected"} @@ -489,25 +477,6 @@ async def _send(self, method: str, params: dict, timeout: float = 300) -> dict: return last_result - def _build_url(self, endpoint_key: str, **kwargs) -> str: - """Build full API URL.""" - path = ENDPOINTS[endpoint_key].format(**kwargs) - sep = "&" if "?" in path else "?" - return f"{GOOGLE_FLOW_API}{path}{sep}key={GOOGLE_API_KEY}" - - def _client_context(self, project_id: str, user_paygate_tier: str = "PAYGATE_TIER_TWO") -> dict: - """Build clientContext with recaptcha placeholder.""" - return { - "projectId": str(project_id), - "recaptchaContext": { - "applicationType": "RECAPTCHA_APPLICATION_TYPE_WEB", - "token": "", # Extension injects real token - }, - "sessionId": f";{int(time.time() * 1000)}", - "tool": "PINHOLE", - "userPaygateTier": user_paygate_tier, - } - # ─── batchexecute transport ────────────────────────────── # # Flow's rewritten frontend signs every call with the session cookie plus a @@ -598,8 +567,6 @@ def flow_project_id(self, requested: str | None = None) -> str | None: return FLOW_PROJECT_ID or None async def create_project(self, project_title: str, tool_name: str = "PINHOLE") -> dict: - if not USE_BATCH_RPC: - return await self._legacy_create_project(project_title, tool_name) pid = self.flow_project_id() if not pid: return {"error": _UNSUPPORTED_CREATE_PROJECT} @@ -621,9 +588,6 @@ async def generate_images(self, prompt: str, project_id: str, old REST one so the parsers downstream do not have to care which transport produced it. """ - if not USE_BATCH_RPC: - return await self._legacy_generate_images( - prompt, project_id, aspect_ratio, user_paygate_tier, character_media_ids) try: if not isinstance(count, int) or isinstance(count, bool) or not 1 <= count <= 4: @@ -726,10 +690,6 @@ async def edit_image(self, prompt: str, source_media_id: str, generic reference conditions a fresh generation; BASE_IMAGE is the wire shape the current Flow editor uses for an actual image edit/refine. """ - if not USE_BATCH_RPC: - return await self._legacy_edit_image( - prompt, source_media_id, project_id, aspect_ratio, - user_paygate_tier, character_media_ids) refs = [mid for mid in (character_media_ids or []) if mid != source_media_id] return await self.generate_images( @@ -747,8 +707,6 @@ async def edit_image(self, prompt: str, source_media_id: str, async def upscale_image(self, media_id: str, project_id: str, resolution: str = "2K") -> dict: """Return Flow's synchronous 2K/4K image upscale as base64 JPEG data.""" - if not USE_BATCH_RPC: - return {"status": 400, "error": "Image upscale requires the flow.google.com batch transport"} try: pid = self._batch_project_id(project_id) freq = fb.image_upscale_request(media_id, resolution) @@ -778,10 +736,6 @@ async def generate_video(self, start_image_media_id: str, prompt: str, end_image_media_id: str = None, user_paygate_tier: str = "PAYGATE_TIER_TWO") -> dict: """Submit an i2v generation. Returns operations for the poller.""" - if not USE_BATCH_RPC: - return await self._legacy_generate_video( - start_image_media_id, prompt, project_id, scene_id, - aspect_ratio, end_image_media_id, user_paygate_tier) if end_image_media_id: if not FLOW_ALLOW_DEGRADED: @@ -815,10 +769,6 @@ async def generate_video_from_references(self, reference_media_ids: list[str], aspect_ratio: str = "VIDEO_ASPECT_RATIO_PORTRAIT", user_paygate_tier: str = "PAYGATE_TIER_TWO") -> dict: """Generate video from multiple reference images (r2v).""" - if not USE_BATCH_RPC: - return await self._legacy_generate_video_from_references( - reference_media_ids, prompt, project_id, scene_id, - aspect_ratio, user_paygate_tier) if not FLOW_ALLOW_DEGRADED: return {"error": _unsupported( @@ -841,8 +791,6 @@ 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.""" - 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", @@ -865,8 +813,6 @@ async def check_video_status(self, operations: list[dict]) -> dict: Everything short of that is PENDING, and the caller's own poll loop owns the timeout. """ - if not USE_BATCH_RPC: - return await self._legacy_check_video_status(operations) out = [] for entry in operations or []: @@ -986,8 +932,6 @@ async def get_credits(self) -> dict: fixed — so on the batch path this answers with the configured default rather than pretending to know. """ - if not USE_BATCH_RPC: - return await self._legacy_get_credits() return {"status": 200, "data": { "userPaygateTier": DEFAULT_PAYGATE_TIER, "note": "batchexecute path: tier is configured (DEFAULT_PAYGATE_TIER), not fetched", @@ -1001,8 +945,6 @@ async def validate_media_id(self, media_id: str) -> bool: async def get_media(self, media_id: str) -> dict: """Fetch a media record, which is where a fresh signed url lives.""" - if not USE_BATCH_RPC: - return await self._legacy_get_media(media_id) try: urls = await self._batch_media_urls(media_id) except Exception as e: @@ -1019,8 +961,6 @@ async def get_media(self, media_id: str) -> dict: 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.""" - if not USE_BATCH_RPC: - return await self._legacy_upload_image(image_base64, mime_type, project_id, file_name) try: pid = self._batch_project_id(project_id) payload = await self._batch_payload( @@ -1033,317 +973,6 @@ async def upload_image(self, image_base64: str, mime_type: str = "image/jpeg", return _batch_error(e) return {"status": 200, "data": {"media": {"name": media_id}}, "_mediaId": media_id} - # ─── Legacy REST methods (aisandbox-pa, pre-migration) ─── - - async def _legacy_create_project(self, project_title: str, tool_name: str = "PINHOLE") -> dict: - """Create a project on Google Flow via tRPC endpoint. - - Returns the full response including projectId. - """ - url = "https://labs.google/fx/api/trpc/project.createProject" - body = {"json": {"projectTitle": project_title, "toolName": tool_name}} - - return await self._send("trpc_request", { - "url": url, - "method": "POST", - "headers": { - "content-type": "application/json", - "accept": "*/*", - }, - "body": body, - }, timeout=30) - - async def _legacy_generate_images(self, prompt: str, project_id: str, - aspect_ratio: str = "IMAGE_ASPECT_RATIO_PORTRAIT", - user_paygate_tier: str = "PAYGATE_TIER_TWO", - character_media_ids: list[str] = None) -> dict: - """Generate image(s). - - If character_media_ids is provided, uses edit_image flow (batchGenerateImages - with imageInputs) — same endpoint, but includes character references. - Without characters, uses plain generate_images. - - Response structure: - data.media[].name = mediaId (used for video gen) - """ - ts = int(time.time() * 1000) - ctx = self._client_context(project_id, user_paygate_tier) - - request_item = { - "clientContext": {**ctx, "sessionId": f";{ts}"}, - "seed": ts % 1000000, - "structuredPrompt": {"parts": [{"text": prompt}]}, - "imageAspectRatio": aspect_ratio, - "imageModelName": IMAGE_MODELS["NANO_BANANA_PRO"], - } - - # Add character references if provided (edit_image flow) - if character_media_ids: - request_item["imageInputs"] = [ - {"name": mid, "imageInputType": "IMAGE_INPUT_TYPE_REFERENCE"} - for mid in character_media_ids - ] - - batch_id = f"{uuid.uuid4()}" if character_media_ids else None - body = { - "clientContext": ctx, - "requests": [request_item], - } - if batch_id: - body["mediaGenerationContext"] = {"batchId": batch_id} - body["useNewMedia"] = True - - url = self._build_url("generate_images", project_id=project_id) - return await self._send("api_request", { - "url": url, - "method": "POST", - "headers": random_headers(), - "body": body, - "captchaAction": "IMAGE_GENERATION", - }) - - async def _legacy_edit_image(self, prompt: str, source_media_id: str, - project_id: str, - aspect_ratio: str = "IMAGE_ASPECT_RATIO_PORTRAIT", - user_paygate_tier: str = "PAYGATE_TIER_ONE", - character_media_ids: list[str] = None) -> dict: - """Edit an existing image using IMAGE_INPUT_TYPE_BASE_IMAGE. - - If character_media_ids is provided, appends them as IMAGE_INPUT_TYPE_REFERENCE - after the base image. Order: [base_image, char_A, char_B, ...]. - This helps Google Flow detect characters for consistent edits. - """ - ts = int(time.time() * 1000) - ctx = self._client_context(project_id, user_paygate_tier) - - image_inputs = [ - {"name": source_media_id, "imageInputType": "IMAGE_INPUT_TYPE_BASE_IMAGE"} - ] - if character_media_ids: - for mid in character_media_ids: - image_inputs.append({"name": mid, "imageInputType": "IMAGE_INPUT_TYPE_REFERENCE"}) - - request_item = { - "clientContext": {**ctx, "sessionId": f";{ts}"}, - "seed": ts % 1000000, - "structuredPrompt": {"parts": [{"text": prompt}]}, - "imageAspectRatio": aspect_ratio, - "imageModelName": IMAGE_MODELS["NANO_BANANA_PRO"], - "imageInputs": image_inputs, - } - - body = { - "clientContext": ctx, - "mediaGenerationContext": {"batchId": f"{uuid.uuid4()}"}, - "useNewMedia": True, - "requests": [request_item], - } - - url = self._build_url("generate_images", project_id=project_id) - return await self._send("api_request", { - "url": url, - "method": "POST", - "headers": random_headers(), - "body": body, - "captchaAction": "IMAGE_GENERATION", - }) - - async def _legacy_generate_video(self, start_image_media_id: str, prompt: str, - project_id: str, scene_id: str, - aspect_ratio: str = "VIDEO_ASPECT_RATIO_PORTRAIT", - end_image_media_id: str = None, - user_paygate_tier: str = "PAYGATE_TIER_TWO") -> dict: - """Generate video from start image (i2v). - - Two sub-types: - - frame_2_video (i2v): startImage only - - start_end_frame_2_video (i2v_fl): startImage + endImage (for scene chaining) - """ - gen_type = "start_end_frame_2_video" if end_image_media_id else "frame_2_video" - model_key = VIDEO_MODELS.get(user_paygate_tier, {}).get(gen_type, {}).get(aspect_ratio) - - if not model_key: - return {"error": f"No model for tier={user_paygate_tier} type={gen_type} ratio={aspect_ratio}"} - - request = { - "aspectRatio": aspect_ratio, - "seed": int(time.time()) % 10000, - "textInput": {"structuredPrompt": {"parts": [{"text": prompt}]}}, - "videoModelKey": model_key, - "startImage": {"mediaId": start_image_media_id}, - "metadata": {"sceneId": scene_id}, - } - - if end_image_media_id: - request["endImage"] = {"mediaId": end_image_media_id} - - endpoint_key = "generate_video_start_end" if end_image_media_id else "generate_video" - body = { - "mediaGenerationContext": {"batchId": f"{uuid.uuid4()}"}, - "clientContext": self._client_context(project_id, user_paygate_tier), - "requests": [request], - "useV2ModelConfig": True, - } - - url = self._build_url(endpoint_key) - return await self._send("api_request", { - "url": url, - "method": "POST", - "headers": random_headers(), - "body": body, - "captchaAction": "VIDEO_GENERATION", - }, timeout=60) # Submit only — polling is separate - - async def _legacy_generate_video_from_references(self, reference_media_ids: list[str], - prompt: str, project_id: str, scene_id: str, - aspect_ratio: str = "VIDEO_ASPECT_RATIO_PORTRAIT", - user_paygate_tier: str = "PAYGATE_TIER_TWO") -> dict: - """Generate video from multiple reference images (r2v). - - Uses referenceImages instead of startImage — the model composes - a video from all provided reference character images. - - Args: - reference_media_ids: List of character media_ids (from uploadImage) - """ - gen_type = "reference_frame_2_video" - model_key = VIDEO_MODELS.get(user_paygate_tier, {}).get(gen_type, {}).get(aspect_ratio) - - if not model_key: - return {"error": f"No model for tier={user_paygate_tier} type={gen_type} ratio={aspect_ratio}"} - - request = { - "aspectRatio": aspect_ratio, - "seed": int(time.time()) % 10000, - "textInput": {"structuredPrompt": {"parts": [{"text": prompt}]}}, - "videoModelKey": model_key, - "referenceImages": [ - {"mediaId": mid, "imageUsageType": "IMAGE_USAGE_TYPE_ASSET"} - for mid in reference_media_ids - ], - "metadata": {}, - } - - body = { - "mediaGenerationContext": {"batchId": f"{uuid.uuid4()}"}, - "clientContext": self._client_context(project_id, user_paygate_tier), - "requests": [request], - "useV2ModelConfig": True, - } - - url = self._build_url("generate_video_references") - return await self._send("api_request", { - "url": url, - "method": "POST", - "headers": random_headers(), - "body": body, - "captchaAction": "VIDEO_GENERATION", - }, timeout=60) - - async def _legacy_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.""" - model_key = UPSCALE_MODELS.get(resolution, "veo_3_1_upsampler_4k") - - body = { - "clientContext": { - "sessionId": f";{int(time.time() * 1000)}", - "recaptchaContext": { - "applicationType": "RECAPTCHA_APPLICATION_TYPE_WEB", - "token": "", - }, - }, - "requests": [{ - "aspectRatio": aspect_ratio, - "resolution": resolution, - "seed": int(time.time()) % 100000, - "metadata": {"sceneId": scene_id}, - "videoInput": {"mediaId": media_id}, - "videoModelKey": model_key, - }], - } - - url = self._build_url("upscale_video") - return await self._send("api_request", { - "url": url, - "method": "POST", - "headers": random_headers(), - "body": body, - "captchaAction": "VIDEO_GENERATION", - }, timeout=60) - - async def _legacy_check_video_status(self, operations: list[dict]) -> dict: - """Check status of video generation operations.""" - body = {"operations": operations} - url = self._build_url("check_video_status") - return await self._send("api_request", { - "url": url, - "method": "POST", - "headers": random_headers(), - "body": body, - }, timeout=30) # No captcha needed - - async def _legacy_get_credits(self) -> dict: - """Get user credits and tier.""" - url = self._build_url("get_credits") - return await self._send("api_request", { - "url": url, - "method": "GET", - "headers": random_headers(), - }, timeout=15) - - async def _legacy_get_media(self, media_id: str) -> dict: - """Fetch media metadata from Google Flow. - - Returns the raw API response which contains a fresh signed URL - in data.fifeUrl or data.servingUri. - """ - url = f"{GOOGLE_FLOW_API}/v1/media/{media_id}?key={GOOGLE_API_KEY}&clientContext.tool=PINHOLE" - return await self._send("api_request", { - "url": url, - "method": "GET", - "headers": random_headers(), - }, timeout=15) - - async def _legacy_upload_image(self, image_base64: str, mime_type: str = "image/jpeg", - project_id: str = "", file_name: str = "image.jpg") -> dict: - """Upload an image for use as start/end frame. - - Uses /v1/flow/uploadImage endpoint. - Response: {media: {name: "uuid", ...}, workflow: {...}} - We store media.name as the mediaId for video generation. - """ - body = { - "clientContext": { - "projectId": project_id, - "tool": "PINHOLE", - }, - "fileName": file_name, - "imageBytes": image_base64, - "isHidden": False, - "isUserUploaded": True, - "mimeType": mime_type, - } - - url = self._build_url("upload_image") - result = await self._send("api_request", { - "url": url, - "method": "POST", - "headers": random_headers(), - "body": body, - }, timeout=60) - - # Extract media.name for convenience (used as mediaId in video gen) - if not _is_ws_error(result): - data = result.get("data", {}) - if isinstance(data, dict): - media = data.get("media", {}) - if isinstance(media, dict) and media.get("name"): - result["_mediaId"] = media["name"] - - return result - # ─── Response shaping ──────────────────────────────────────── # # The batch path answers in Flow's positional arrays; everything downstream diff --git a/agent/services/headers.py b/agent/services/headers.py deleted file mode 100644 index b7e1ac39c..000000000 --- a/agent/services/headers.py +++ /dev/null @@ -1,40 +0,0 @@ -"""Randomized browser headers to avoid detection.""" -import random -import time -from agent.config import USER_AGENTS, CHROME_VERSIONS, BROWSER_VALIDATIONS, CLIENT_DATA - - -def random_headers() -> dict: - """Generate randomized browser-like headers for Google Flow API.""" - ua = random.choice(USER_AGENTS) - chrome_ver = random.choice(CHROME_VERSIONS) - validation = random.choice(BROWSER_VALIDATIONS) - client_data = random.choice(CLIENT_DATA) - - if "Windows" in ua: - platform, mobile = '"Windows"', "?0" - elif "Macintosh" in ua: - platform, mobile = '"macOS"', "?0" - else: - platform, mobile = '"Linux"', "?0" - - return { - "accept": "*/*", - "accept-language": "en-US,en;q=0.9", - "content-type": "text/plain;charset=UTF-8", - "origin": "https://labs.google", - "priority": "u=1, i", - "referer": "https://labs.google/", - "sec-ch-ua": f'"Not;A=Brand";v="99", {chrome_ver}', - "sec-ch-ua-mobile": mobile, - "sec-ch-ua-platform": platform, - "sec-fetch-dest": "empty", - "sec-fetch-mode": "cors", - "sec-fetch-site": "cross-site", - "user-agent": ua, - "x-browser-channel": "stable", - "x-browser-copyright": "Copyright 2025 Google LLC. All rights reserved.", - "x-browser-validation": validation, - "x-browser-year": "2025", - "x-client-data": client_data, - } diff --git a/agent/services/omni_flash.py b/agent/services/omni_flash.py index 796d6a312..8de784447 100644 --- a/agent/services/omni_flash.py +++ b/agent/services/omni_flash.py @@ -1,44 +1,34 @@ """Gemini Omni Flash video generation through the Google Flow bridge. -Supported Omni surfaces in this module: - -* first frame -> video via ``batchAsyncGenerateVideoStartImage`` -* first + last frame -> video via ``batchAsyncGenerateVideoStartAndEndImage`` -* reference images -> video via ``batchAsyncGenerateVideoReferenceImages`` - -Omni duration-specific model keys live in ``agent/models.json``. First-frame -Flow requests have been captured with ``abra_i2v_s``. The current -First+Last rollout uses the same Omni I2V family but the StartAndEnd endpoint; -that mapping is deliberately configurable separately so it can be changed -without a code release if Google's rollout rotates the wire key. - -Important: Omni submit responses may contain operation-looking handles, but -those handles are not compatible with the legacy -``batchCheckAsyncVideoGenerationStatus`` polling endpoint. Omni jobs are -workflow-backed and are polled through Flow's authenticated project data. +What works today: **text-to-video**, on flow.google.com batchexecute. It submits +through ``fb.text_video_request`` like every other migrated call. + +What does not: first-frame, first+last, and reference-to-video. Those three were +only ever implemented against the pre-migration REST endpoints, and that +transport was removed when Flow stopped minting the bearer it needed. Their +migrated payloads have not been captured, so they validate their inputs and then +fail by name — see ``docs/CAPTURE.md`` for how to record one off the new UI. + +Duration-specific model keys live in ``agent/models.json`` so a rollout that +rotates a wire key does not need a code release. + +Polling is workflow-backed: a submit hands back ``name`` + ``primaryMediaId``, +not an operation handle, so Omni jobs must not be fed to ``check_video_status``. """ from __future__ import annotations import json -import time -import uuid from pathlib import Path -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 _MODELS_FILE = Path(__file__).parent.parent / "models.json" -#: Every Omni surface here rides the pre-migration transports — the REST -#: endpoints on aisandbox-pa and the labs.google tRPC snapshot it polls -#: through. Flow moved to flow.google.com in September 2026 and stopped -#: minting the bearer both of those need, and no Omni payload has been -#: captured off the new frontend, so on the batch path these fail with a -#: name rather than dying on a 401 five retries deep. +#: Frame and reference generation had only a REST implementation, and that +#: transport is gone. Failing by name here beats dying on a 401 five retries +#: deep, and names the thing a capture would restore. _UNSUPPORTED_ON_BATCH = ( "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 " @@ -47,10 +37,6 @@ ) -def _batch_path_blocks_omni() -> dict | None: - """The error to return instead of reaching for auth that is gone.""" - return {"error": _UNSUPPORTED_ON_BATCH} if USE_BATCH_RPC else None - OMNI_FLASH_VALID_DURATIONS = (4, 6, 8, 10) OMNI_FLASH_VALID_ASPECTS = { "VIDEO_ASPECT_RATIO_PORTRAIT", @@ -61,42 +47,6 @@ def _batch_path_blocks_omni() -> dict | None: OMNI_FLASH_CREDIT_COST = {4: 15, 6: 20, 8: 25, 10: 30} -async def _fetch_project_initial_data(client, project_id: str) -> dict: - """Fetch the same authenticated project snapshot used by the Flow UI.""" - query = quote( - json.dumps({"json": {"projectId": project_id}}, separators=(",", ":")), - safe="", - ) - url = f"https://labs.google/fx/api/trpc/flow.projectInitialData?input={query}" - return await client._send( - "trpc_request", - { - "url": url, - "method": "GET", - "headers": {"content-type": "application/json"}, - }, - timeout=15, - ) - - -async def _fetch_media_url(client, media_id: str) -> dict: - """Resolve Flow's authenticated media redirect without buffering the file.""" - url = ( - "https://labs.google/fx/api/trpc/media.getMediaUrlRedirect" - f"?name={quote(media_id, safe='')}" - ) - return await client._send( - "trpc_request", - { - "url": url, - "method": "GET", - "headers": {"content-type": "application/json"}, - "responseMode": "url", - }, - timeout=15, - ) - - def _validate_duration(duration_s: int) -> None: if duration_s not in OMNI_FLASH_VALID_DURATIONS: raise ValueError( @@ -201,23 +151,6 @@ def extract_omni_workflows(result: dict) -> list[dict]: return normalized -def _annotate_polling(result: dict, project_id: str) -> dict: - """Add an explicit FlowKit polling descriptor to a successful submit.""" - workflows = extract_omni_workflows(result) - if not workflows: - return result - data = result.get("data") if isinstance(result.get("data"), dict) else result - if isinstance(data, dict): - for workflow in workflows: - workflow["project_id"] = project_id - data["flowkitPolling"] = { - "mode": "project_media", - "project_id": project_id, - "workflows": workflows, - } - return result - - async def generate_omni_flash_text_video( prompt: str, project_id: str, @@ -230,9 +163,6 @@ async def generate_omni_flash_text_video( """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) @@ -279,61 +209,13 @@ async def _submit_omni_frame_video( seed: int | None = None, ) -> dict: """Submit Omni first-frame or First+Last generation.""" - blocked = _batch_path_blocks_omni() - if blocked: - return blocked _validate_frame_inputs( start_image_media_id, end_image_media_id, duration_s, aspect_ratio, ) - - mode = ( - "start_end_frame_to_video" - if end_image_media_id is not None - else "frame_to_video" - ) - endpoint = ( - "generate_video_start_end" - if end_image_media_id is not None - else "generate_video" - ) - model_key = _load_model_key(duration_s, mode=mode) - client = get_flow_client() - ts = int(time.time() * 1000) - - request_item = { - "aspectRatio": aspect_ratio, - "textInput": {"structuredPrompt": {"parts": [{"text": prompt}]}}, - "videoModelKey": model_key, - "seed": seed if seed is not None else ts % 1_000_000, - "metadata": {"sceneId": scene_id} if scene_id else {}, - "startImage": {"mediaId": start_image_media_id}, - } - if end_image_media_id is not None: - request_item["endImage"] = {"mediaId": end_image_media_id} - - context = client._client_context(project_id, user_paygate_tier) - body = { - "mediaGenerationContext": {"batchId": str(uuid.uuid4())}, - "clientContext": {**context, "sessionId": f";{ts}"}, - "requests": [request_item], - "useV2ModelConfig": True, - } - - result = await client._send( - "api_request", - { - "url": client._build_url(endpoint), - "method": "POST", - "headers": random_headers(), - "body": body, - "captchaAction": "VIDEO_GENERATION", - }, - timeout=60, - ) - return _annotate_polling(result, project_id) + return {"error": _UNSUPPORTED_ON_BATCH} async def generate_omni_flash_first_frame_video( @@ -401,50 +283,8 @@ async def generate_omni_flash_video( the workflow names and primary media IDs required by the Omni polling path. Do not feed Omni operation handles to ``check_video_status``. """ - blocked = _batch_path_blocks_omni() - if blocked: - return blocked - refs = _validate_reference_inputs(reference_media_ids, duration_s, aspect_ratio) - model_key = _load_model_key(duration_s, mode="reference_to_video") - client = get_flow_client() - - ts = int(time.time() * 1000) - request_item = { - "aspectRatio": aspect_ratio, - "textInput": {"structuredPrompt": {"parts": [{"text": prompt}]}}, - "videoModelKey": model_key, - "seed": seed if seed is not None else ts % 1_000_000, - "metadata": {"sceneId": scene_id} if scene_id else {}, - "referenceImages": [ - {"mediaId": mid, "imageUsageType": "IMAGE_USAGE_TYPE_ASSET"} - for mid in refs - ], - } - - context = client._client_context(project_id, user_paygate_tier) - body = { - "mediaGenerationContext": { - "batchId": str(uuid.uuid4()), - "audioFailurePreference": "BLOCK_SILENCED_VIDEOS", - }, - "clientContext": {**context, "sessionId": f";{ts}"}, - "requests": [request_item], - "useV2ModelConfig": True, - } - - url = client._build_url("generate_video_references") - result = await client._send( - "api_request", - { - "url": url, - "method": "POST", - "headers": random_headers(), - "body": body, - "captchaAction": "VIDEO_GENERATION", - }, - timeout=60, - ) - return _annotate_polling(result, project_id) + _validate_reference_inputs(reference_media_ids, duration_s, aspect_ratio) + return {"error": _UNSUPPORTED_ON_BATCH} async def _check_omni_batch_media( @@ -507,166 +347,4 @@ async def check_omni_flash_status( 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) - if item: - normalized.append(item) - if not normalized: - raise ValueError( - "Omni polling requires workflow descriptors with name and primary_media_id " - "(or raw Flow metadata.primaryMediaId)" - ) - - resolved_project_id = project_id or next( - (item.get("project_id", "") for item in normalized if item.get("project_id")), - "", - ) - if not resolved_project_id: - raise ValueError( - "Omni project polling requires project_id. Use the project_id returned " - "inside flowkitPolling or pass project_id explicitly." - ) - if any( - item.get("project_id") and item["project_id"] != resolved_project_id - for item in normalized - ): - raise ValueError( - "All Omni workflows in one poll must belong to the same project_id" - ) - - client = get_flow_client() - response = await _fetch_project_initial_data(client, resolved_project_id) - http_status = response.get("status") if isinstance(response, dict) else None - if isinstance(http_status, int) and http_status >= 400: - data = response.get("data") if isinstance(response.get("data"), dict) else {} - error = data.get("error") if isinstance(data, dict) else None - if isinstance(error, dict): - error = error.get("message") or error.get("code") - raise RuntimeError( - error - or response.get("error") - or f"Flow project poll failed: API_{http_status}" - ) - - envelope = response.get("data") if isinstance(response, dict) else None - result = envelope.get("result") if isinstance(envelope, dict) else None - result_data = result.get("data") if isinstance(result, dict) else None - project_json = result_data.get("json") if isinstance(result_data, dict) else None - contents = project_json.get("projectContents") if isinstance(project_json, dict) else None - if not isinstance(contents, dict): - raise RuntimeError("Flow project poll returned an unexpected response shape") - - project_workflows = contents.get("workflows") - project_media = contents.get("media") - project_workflows = project_workflows if isinstance(project_workflows, list) else [] - project_media = project_media if isinstance(project_media, list) else [] - known_workflow_names = { - item.get("name") - for item in project_workflows - if isinstance(item, dict) and isinstance(item.get("name"), str) - } - media_by_id = { - item.get("name"): item - for item in project_media - if isinstance(item, dict) and isinstance(item.get("name"), str) - } - media_by_workflow = { - item.get("workflowId"): item - for item in project_media - if isinstance(item, dict) and isinstance(item.get("workflowId"), str) - } - - items = [] - - for workflow in normalized: - name = workflow["name"] - media_id = workflow["primary_media_id"] - payload = media_by_id.get(media_id) or media_by_workflow.get(name) - if not isinstance(payload, dict): - items.append({ - "name": name, - "primary_media_id": media_id, - "project_id": resolved_project_id, - "done": False, - "status": "PENDING", - "error": None, - "workflow_present": name in known_workflow_names, - }) - continue - - metadata = payload.get("mediaMetadata") - metadata = metadata if isinstance(metadata, dict) else {} - media_status = metadata.get("mediaStatus") - media_status = media_status if isinstance(media_status, dict) else {} - generation_status = media_status.get("mediaGenerationStatus") - - if isinstance(generation_status, str) and ( - generation_status.endswith("FAILED") or generation_status.endswith("CANCELLED") - ): - items.append({ - "name": name, - "primary_media_id": media_id, - "project_id": resolved_project_id, - "done": True, - "status": "FAILED", - "error": generation_status, - }) - continue - - if generation_status != "MEDIA_GENERATION_STATUS_SUCCESSFUL": - items.append({ - "name": name, - "primary_media_id": media_id, - "project_id": resolved_project_id, - "done": False, - "status": "PENDING", - "error": None, - }) - continue - - url = None - url_error = None - url_response = await _fetch_media_url(client, media_id) - if isinstance(url_response, dict) and url_response.get("status", 500) < 400: - url_data = url_response.get("data") - candidate = url_data.get("url") if isinstance(url_data, dict) else None - if isinstance(candidate, str) and candidate.startswith("https://flow-content.google/"): - url = candidate - else: - url_error = "Flow media redirect returned no allowed URL" - else: - url_error = ( - url_response.get("error") - if isinstance(url_response, dict) - else "Flow media redirect failed" - ) - item = { - "name": name, - "primary_media_id": media_id, - "project_id": resolved_project_id, - "done": True, - "status": "MEDIA_GENERATION_STATUS_SUCCESSFUL", - "error": None, - "media": { - "media_id": media_id, - "url": url, - "encoded_video_available": False, - }, - } - if include_encoded_video: - item["media"]["encoded_video"] = None - if url_error: - item["media"]["url_error"] = url_error - items.append(item) - - all_done = bool(items) and all(item["done"] for item in items) - any_failed = any(item.get("status") == "FAILED" for item in items) - return { - "project_id": resolved_project_id, - "done": all_done, - "status": "FAILED" if any_failed else ("COMPLETED" if all_done else "PENDING"), - "workflows": items, - } + return await _check_omni_batch_media(workflows, include_encoded_video, project_id) diff --git a/docs/OMNI_FLASH.md b/docs/OMNI_FLASH.md index 64d3c040d..bf8a2e607 100644 --- a/docs/OMNI_FLASH.md +++ b/docs/OMNI_FLASH.md @@ -22,7 +22,7 @@ Use `http://127.0.0.1:8100` when the caller runs on the FlowKit host. For a remo ## Supported modes -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`. +On `flow.google.com`, Omni **text-to-video** is migrated and live-verified. Frame and reference generation had only a REST implementation; that transport has been removed, so they validate their inputs and then fail by name with `UNSUPPORTED_ON_BATCH_API` until their payloads are captured (see `docs/CAPTURE.md`). | Mode | Batch status | Endpoint | Internal model family | |---|---|---|---| diff --git a/extension/background.js b/extension/background.js index 79036f87b..6def15cfd 100644 --- a/extension/background.js +++ b/extension/background.js @@ -8,8 +8,10 @@ * `Bearer ya29.…` the old REST host needed. The current path is `batch_rpc`: * the agent builds an `f.req` envelope, this worker mints a captcha for it and * runs the POST in the page's MAIN world, where the `at` CSRF token lives. - * The bearer capture and `api_request` proxy below are the legacy path, kept - * for USE_BATCH_RPC=0 and for an old pinned labs.google tab. + * The bearer capture and the `api_request` / `trpc_request` proxies below are + * the pre-migration path. The agent no longer sends either — it speaks only + * `batch_rpc`. They stay so an extension updated ahead of its agent keeps + * serving an older one; remove them once no agent in the wild sends them. */ const AGENT_WS_URL = 'ws://127.0.0.1:9222'; @@ -643,8 +645,12 @@ async function handleTrpcRequest(msg) { } } -// Legacy REST proxy against aisandbox-pa. Reachable only with USE_BATCH_RPC=0 -// on a profile that still holds a `Bearer ya29.…`; Flow stopped minting those. +// Legacy REST proxy against aisandbox-pa. No current agent sends `api_request`; +// kept only so an extension updated ahead of its agent still serves an older +// one. It needs a `Bearer ya29.…` that Flow stopped minting, so it 401s on any +// post-migration profile — as does sendTelemetry below, which early-returns +// without a flowKey. Nothing here reaches aisandbox-pa any more; when the +// oldest agent in the wild speaks batch_rpc, this and the host permission go. async function handleApiRequest(msg) { const { id, params } = msg; const { url, method, headers, body, captchaAction } = params; diff --git a/extension/manifest.json b/extension/manifest.json index a01b657e2..78eebc2d8 100644 --- a/extension/manifest.json +++ b/extension/manifest.json @@ -1,7 +1,7 @@ { "manifest_version": 3, "name": "Flow Kit", - "version": "0.3.1", + "version": "0.3.2", "description": "Local agent bridge for Google Flow \u2014 runs batchexecute RPCs inside a signed-in flow.google.com tab, mints reCAPTCHA, proxies API calls", "permissions": [ "storage", diff --git a/skills/fk-doctor.md b/skills/fk-doctor.md index 6b92019c1..a13074b9c 100644 --- a/skills/fk-doctor.md +++ b/skills/fk-doctor.md @@ -6,7 +6,7 @@ Diagnose any FlowKit error and prescribe a fix. Knows the full error taxonomy ac - Any `/api/requests/*` response has `status=FAILED` or `error_message` is set - A request has been `PROCESSING` for > 10 minutes with no progress - `GET /health` returns `extension_connected: false` -- User reports any error string containing: `UNSAFE_GENERATION`, `QUOTA`, `not found`, `CAPTCHA`, `UNUSUAL_ACTIVITY`, `NO_AT_TOKEN`, `NO_FLOW_PROJECT`, `UNSUPPORTED_ON_BATCH_API`, `NO_FLOW_KEY`, `NO_FLOW_TAB`, `FLOW_TAB_DISCARDED`, `extension_switched`, `Failed to fetch`, `MODEL_ACCESS_DENIED`, `PAYGATE_TIER_TWO`, `invalidTags`, `quotaExceeded`, `invalid_grant` +- User reports any error string containing: `UNSAFE_GENERATION`, `QUOTA`, `not found`, `CAPTCHA`, `UNUSUAL_ACTIVITY`, `NO_AT_TOKEN`, `NO_FLOW_PROJECT`, `UNSUPPORTED_ON_BATCH_API`, `NO_FLOW_TAB`, `FLOW_TAB_DISCARDED`, `NO_INJECTION_RESULT`, `extension_switched`, `Failed to fetch`, `MODEL_ACCESS_DENIED`, `PAYGATE_TIER_TWO`, `invalidTags`, `quotaExceeded`, `invalid_grant` - User asks "why did X fail", "what's wrong with the pipeline", "why is this stuck", "tại sao X lỗi", "lỗi gì vậy" - An HTTP 4xx/5xx reaches the main agent from any endpoint under `127.0.0.1:8100` - A YouTube upload returns `HttpError` from `googleapiclient` @@ -67,25 +67,23 @@ Cross-reference `error_message` against the taxonomy below. Print: **Diagnosis / Match against taxonomy — even partial matches (`"not found"`, `"captcha"`, `"quota"`). -## Which transport is running +## The transport -Since Flow moved to `flow.google.com` (September 2026) there are two paths, and -the taxonomy below splits on which one is live: +One path since Flow moved to `flow.google.com` (September 2026): ```bash -python3 -c "from agent.config import USE_BATCH_RPC, FLOW_PROJECT_ID; \ - print('batch' if USE_BATCH_RPC else 'legacy REST', '| project:', FLOW_PROJECT_ID or 'UNPINNED')" +python3 -c "from agent.config import FLOW_PROJECT_ID; \ + print('batch | project:', FLOW_PROJECT_ID or 'UNPINNED')" ``` -- **batch** (default) — the agent builds an `f.req` envelope, the extension runs - it inside a signed-in `flow.google.com` tab. No bearer token exists on this - path, so `flow_key_present: false` in `/api/flow/status` is **normal**, not a - fault. Requires a Flow tab open and `FLOW_PROJECT_ID` pinned. -- **legacy REST** (`USE_BATCH_RPC=0`) — the pre-migration `aisandbox-pa` path. - It needs a `Bearer ya29.…` Flow no longer mints, so it will 401 on any fresh - profile. Treat any report of it "suddenly breaking" as the migration, not a - regression: if `token_age_s` only climbs across tab reloads, the token is not - stale, it is gone. +The agent builds an `f.req` envelope and the extension runs it inside a +signed-in `flow.google.com` tab. **No bearer token exists on this path**, so +`flow_key_present: false` in `/api/flow/status` is normal, not a fault. +Requires a Flow tab open and `FLOW_PROJECT_ID` pinned. + +The `aisandbox-pa` REST path that preceded it has been removed — it needed a +`Bearer ya29.…` Flow no longer mints. If an old report mentions it, that is the +migration, not a regression. ## Error Taxonomy @@ -106,13 +104,13 @@ python3 -c "from agent.config import USE_BATCH_RPC, FLOW_PROJECT_ID; \ | Status | Origin | When you see it | |--------|--------|-----------------| | **400** | Flow API | Invalid payload / UNSAFE / entity-not-found — **route by `details.reason`** | -| **401** | Flow API (legacy path only) | Bearer expired — and on a post-migration profile it is not expired, it was never minted. Switch to the batch path | +| **401** | Flow API | Should not happen — batchexecute authenticates in the page, not with a bearer. Check the Flow tab is signed in; see `NO_AT_TOKEN` | | **403** | Extension (`background.js:432`) | `CAPTCHA_FAILED`, `NO_FLOW_TAB`, or `MODEL_ACCESS_DENIED` — read the suffix | | **404** | Flow API | `media_id` not found — same handler as "entity not found" | | **429** | Flow API | Rate-limit or quota — backoff; if message mentions QUOTA_REACHED, terminal | | **500** | Flow backend **or** extension fetch exception (`background.js:504`) | Transient — retry with backoff | | **502** | FastAPI (`agent/api/flow.py:80,92`) | Extension returned error without explicit status — treat as transient | -| **503** | FastAPI | "Extension not connected" or `NO_FLOW_KEY` — worker re-queues PENDING, waits | +| **503** | FastAPI | "Extension not connected" — worker re-queues PENDING, waits | | **504** | Agent | 60s WS timeout waiting for extension — transient, re-queue | Detection lives in `agent/worker/_parsing.py:_is_error`. A result is treated as an error if ANY of these hold: @@ -127,7 +125,6 @@ Detection lives in `agent/worker/_parsing.py:_is_error`. A result is treated as | `Extension not connected` | WS dropped or extension offline | Reload extension at `chrome://extensions`; worker auto-retries | | `extension reconnected` / `extension disconnected` | WS bounce mid-request | Auto re-queue, `retry_count` NOT incremented | | `extension_switched` | User switched active Flow tab | Auto re-queue | -| `NO_FLOW_KEY` | No bearer token captured — **legacy path only**; expected and harmless on the batch path | Only meaningful with `USE_BATCH_RPC=0`; otherwise ignore | | `NO_FLOW_TAB` | No Flow tab for CAPTCHA solve or RPC signing | Open `https://flow.google.com/` and sign in | | `Failed to fetch` | Network drop inside service worker | Auto-retry with backoff | | WS 60s timeout | Extension hung | Reload extension; worker re-queues | @@ -139,9 +136,10 @@ Detection lives in `agent/worker/_parsing.py:_is_error`. A result is treated as | `NO_AT_TOKEN` | The Flow tab loaded but `WIZ_global_data.SNlM0e` is absent — the page is signed out, on an interstitial, or still booting | Retried with backoff | Open `https://flow.google.com/`, confirm you are signed in, let the app finish loading | | `NO_FLOW_TAB` | No Flow tab to sign the request | Extension opens one and retries once | Leave one signed-in Flow tab open; nothing here works headless | | `FLOW_TAB_DISCARDED` | Chrome discarded the backgrounded tab and the reload did not revive it | Retried with backoff | Pin the Flow tab, or keep its window visible | +| `NO_INJECTION_RESULT` | `chrome.scripting.executeScript` resolved with no frame result (`background.js:540`), so the envelope never ran — the Flow tab went away or was still booting at the moment of the call. Most likely on the first captcha-bearing RPC after an agent restart | Retried with backoff, counts against `MAX_RETRIES` | Usually transient — retry first. Rule out the lookalikes before digging: a failed mint reports `CAPTCHA_FAILED`, an oversized payload is not it (`MAX_RPC_TEXT` is 32 MB), and a structured-clone failure surfaces as 500, not 502. If it repeats, pin the Flow tab and reload the extension | | `NO_FLOW_PROJECT` | No Flow project to scope the RPC to | **Terminal — not retried** | Create a project in the Flow UI, pin its uuid as `FLOW_PROJECT_ID` (or pass `flow_project_id` on `POST /api/projects`) | | `UNSUPPORTED_ON_BATCH_API` | A capability whose payload was never captured off the new UI: **video upscale**, **r2v**, **start+end-frame chaining** | **Terminal — not retried** | For chaining and r2v, `FLOW_ALLOW_DEGRADED=1` falls back to plain i2v off the start frame. Upscale has no fallback. Real fix: capture the payload — `docs/CAPTURE.md` | -| `UNSUPPORTED_ON_BATCH_API: Omni Flash` | Omni speaks the pre-migration REST + tRPC endpoints; no batchexecute payload captured | **Terminal — not retried** | Use `model_family=veo`, or `USE_BATCH_RPC=0` on a profile that still holds a bearer | +| `UNSUPPORTED_ON_BATCH_API: Omni Flash` | Omni frame/reference had only a REST implementation; no batchexecute payload captured | **Terminal — not retried** | Use `model_family=veo`, or Omni **text-to-video**, which is migrated | | `PUBLIC_ERROR_UNUSUAL_ACTIVITY` | A reCAPTCHA token was replayed — they are single-use | Retried as a captcha error | Usually self-clears; if it persists the extension is reusing a token, reload it | | `no ogiZ0b envelope in response` | The RPC answered but not with the payload we came for — usually a signed-out page returning an HTML redirect | Retried with backoff | Re-sign in on the Flow tab | | `Polling timeout after Ns: Media not found.` | The job never produced media inside the budget | Terminal after `MAX_RETRIES` | The quoted complaint is a **diagnostic, not the cause** — finished jobs report it too. Check the Flow UI: if the clip is there, raise `VIDEO_POLL_TIMEOUT` | @@ -191,7 +189,7 @@ When the user describes a symptom in plain language, map it here first. | Problem | Solution | |---------|----------| | Extension shows "Agent disconnected" | Start `python -m agent.main` | -| Extension shows "No token" | Expected on the batch path — there is no bearer any more. Only act on it with `USE_BATCH_RPC=0` | +| Extension shows "No token" | Expected — there is no bearer any more. Not a fault | | `CAPTCHA_FAILED: NO_FLOW_TAB` | Open `https://flow.google.com/` — and check the extension is v0.3.0+, older builds only matched the dead labs.google URL and could not see the tab that was right there | | 403 `MODEL_ACCESS_DENIED` | Tier mismatch — `GET /api/flow/credits`, downgrade model in `models.json` via `/fk-change-model` | | 403 `PUBLIC_ERROR_UNUSUAL_ACTIVITY` / `reCAPTCHA evaluation failed` | Google flagged the session as bot-like (rapid bursts, VPN/shared IP, stale cookies). **Pause submits**, then in Chrome: `chrome://settings/cookies` → remove cookies for `google.com` and `labs.google` → reload `flow.google.com` → sign in & solve any captcha → resubmit with ≥1s gap and ≤5 concurrent. Switch network or wait 1–6 h if still blocked | @@ -224,7 +222,6 @@ Always end with a prescription block: Symptom: Root cause: Layer: Flow | Extension | FastAPI | Worker | YouTube | Env -Transport: batch (flow.google.com) | legacy REST (aisandbox-pa) Auto-handler: === FIX === diff --git a/skills/fk-refresh-urls.md b/skills/fk-refresh-urls.md index 5c364eb18..9689d56e0 100644 --- a/skills/fk-refresh-urls.md +++ b/skills/fk-refresh-urls.md @@ -116,7 +116,7 @@ curl -X PATCH "http://127.0.0.1:8100/api/scenes/" \ | Issue | Cause | Fix | |-------|-------|-----| -| `flow_key_present: false` | No bearer token — **expected on the batch path** | Ignore; only meaningful with `USE_BATCH_RPC=0` | +| `flow_key_present: false` | No bearer token — **expected**; batchexecute authenticates in the page | Ignore | | `Extension not connected` | Chrome extension WS disconnected | Check the extension is enabled, refresh the Flow tab | | `refreshed: 0`, `found: 0` | No media ids stored for this project | Nothing to refresh — check the project id | | `refreshed: 0`, `found: N` | Every re-sign failed | Read the agent log; usually `NO_FLOW_TAB` or a signed-out Flow tab | diff --git a/tests/unit/test_flow_client_batch.py b/tests/unit/test_flow_client_batch.py index c38c09ed1..f91d61da0 100644 --- a/tests/unit/test_flow_client_batch.py +++ b/tests/unit/test_flow_client_batch.py @@ -32,7 +32,6 @@ def client(monkeypatch): envelope as well as on what came back. """ import agent.services.flow_client as module - monkeypatch.setattr(module, "USE_BATCH_RPC", True) monkeypatch.setattr(module, "FLOW_PROJECT_ID", PROJECT) monkeypatch.setattr(module, "FLOW_ALLOW_DEGRADED", False) diff --git a/tests/unit/test_omni_flash.py b/tests/unit/test_omni_flash.py index cf3b2fb14..617054eb1 100644 --- a/tests/unit/test_omni_flash.py +++ b/tests/unit/test_omni_flash.py @@ -1,9 +1,20 @@ -"""Unit tests for Gemini Omni Flash Flow submissions and workflow polling. - -Omni speaks the pre-migration transports — the REST endpoints on aisandbox-pa -and the labs.google tRPC snapshot it polls through — so the wire contracts -asserted here are legacy-path contracts and the module is pinned to that path -for the file. What happens on the batch path is one test at the bottom. +"""Unit tests for Gemini Omni Flash submissions and workflow polling. + +Omni has one working surface — text-to-video on batchexecute — and three that +are not ported. So these cover three things: `_load_model_key`, the input +validation that runs before any transport is touched, and that the unported +modes say so instead of reaching for auth that is gone. + +Note on `_load_model_key`: it currently has **no production caller**. Frame and +reference generation used to call it and now return early, and text-to-video +hardcodes `abra_t2v_{n}s` instead of reading models.json. The cases below pin +the mapping for when that is fixed; they are not evidence that model keys are +configurable today. + +The REST and tRPC wire-contract tests that used to live here went with the +transport they asserted. They were pinned to the legacy path by a module-level +autouse fixture, which meant most of this file was testing code that no longer +runs. Git history has them if a capture ever needs a reference. """ from unittest.mock import AsyncMock, MagicMock, patch @@ -13,8 +24,6 @@ import agent.services.omni_flash as omni_flash from agent.services.omni_flash import ( OMNI_FLASH_MAX_REFERENCE_IMAGES, - _fetch_media_url, - _fetch_project_initial_data, _load_model_key, check_omni_flash_status, extract_omni_workflows, @@ -25,20 +34,11 @@ ) -@pytest.fixture(autouse=True) -def legacy_transport(monkeypatch): - """Omni is only reachable on the pre-migration path; assert it there.""" - monkeypatch.setattr(omni_flash, "USE_BATCH_RPC", False) - +# ─── model key policy (models.json, transport-independent) ────────────────── @pytest.mark.parametrize( ("duration", "expected"), - [ - (4, "abra_r2v_4s"), - (6, "abra_r2v_6s"), - (8, "abra_r2v_8s"), - (10, "abra_r2v_10s"), - ], + [(4, "abra_r2v_4s"), (6, "abra_r2v_6s"), (8, "abra_r2v_8s"), (10, "abra_r2v_10s")], ) def test_omni_reference_duration_model_keys(duration, expected): assert _load_model_key(duration) == expected @@ -46,12 +46,7 @@ def test_omni_reference_duration_model_keys(duration, expected): @pytest.mark.parametrize( ("duration", "expected"), - [ - (4, "abra_i2v_4s"), - (6, "abra_i2v_6s"), - (8, "abra_i2v_8s"), - (10, "abra_i2v_10s"), - ], + [(4, "abra_i2v_4s"), (6, "abra_i2v_6s"), (8, "abra_i2v_8s"), (10, "abra_i2v_10s")], ) def test_omni_first_frame_model_keys(duration, expected): assert _load_model_key(duration, mode="frame_to_video") == expected @@ -59,14 +54,11 @@ def test_omni_first_frame_model_keys(duration, expected): @pytest.mark.parametrize( ("duration", "expected"), - [ - (4, "abra_i2v_4s"), - (6, "abra_i2v_6s"), - (8, "abra_i2v_8s"), - (10, "abra_i2v_10s"), - ], + [(4, "abra_i2v_4s"), (6, "abra_i2v_6s"), (8, "abra_i2v_8s"), (10, "abra_i2v_10s")], ) def test_omni_first_last_model_keys_are_independently_configured(duration, expected): + """Same family as first-frame today, but its own key so a rollout that + splits them does not need a code release.""" assert _load_model_key(duration, mode="start_end_frame_to_video") == expected @@ -75,21 +67,20 @@ def test_invalid_duration_fails_before_submit(): _load_model_key(5) +# ─── polling descriptors ──────────────────────────────────────────────────── + def test_extract_omni_workflows_uses_primary_media_id(): + """A submit carries operation-looking handles that are NOT pollable as + operations. The workflow's primaryMediaId is the one that is.""" result = { "status": 200, "data": { "operations": [ - { - "operation": {"name": "operation-looking-handle"}, - "status": "MEDIA_GENERATION_STATUS_PENDING", - } + {"operation": {"name": "operation-looking-handle"}, + "status": "MEDIA_GENERATION_STATUS_PENDING"} ], "workflows": [ - { - "name": "workflow-1", - "metadata": {"primaryMediaId": "media-1"}, - } + {"name": "workflow-1", "metadata": {"primaryMediaId": "media-1"}} ], }, } @@ -98,44 +89,28 @@ def test_extract_omni_workflows_uses_primary_media_id(): ] -def _mock_submit_client(): +@pytest.mark.asyncio +async def test_batch_omni_poll_uses_as29s_media(): client = MagicMock() - client._client_context.return_value = { - "projectId": "project-1", - "tool": "PINHOLE", - "userPaygateTier": "PAYGATE_TIER_ONE", - "recaptchaContext": { - "applicationType": "RECAPTCHA_APPLICATION_TYPE_WEB", - "token": "", - }, - "sessionId": ";old", - } - client._build_url.side_effect = lambda name: f"https://example.test/{name}" - client._send = AsyncMock( - return_value={ - "status": 200, - "data": { - "operations": [ - { - "operation": {"name": "op-1"}, - "status": "MEDIA_GENERATION_STATUS_PENDING", - } - ], - "workflows": [ - { - "name": "workflow-1", - "metadata": {"primaryMediaId": "media-1"}, - } - ], - }, - } - ) - return client + 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") +# ─── text-to-video: the one Omni surface on the batch path ────────────────── + @pytest.mark.asyncio -async def test_batch_text_video_builds_4s_yhhmef_submit(monkeypatch): - monkeypatch.setattr(omni_flash, "USE_BATCH_RPC", True) +async def test_batch_text_video_builds_4s_yhhmef_submit(): + import json + client = MagicMock() client._batch_project_id.return_value = "11111111-2222-3333-4444-555555555555" client._batch_payload = AsyncMock(return_value=[ @@ -159,386 +134,69 @@ async def test_batch_text_video_builds_4s_yhhmef_submit(monkeypatch): 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]) + payload = json.loads(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() - - with patch("agent.services.omni_flash.get_flow_client", return_value=client): - result = await generate_omni_flash_first_frame_video( - start_image_media_id="start-1", - prompt="Camera slowly pushes in", - project_id="project-1", - scene_id="scene-1", - duration_s=10, - aspect_ratio="VIDEO_ASPECT_RATIO_LANDSCAPE", - user_paygate_tier="PAYGATE_TIER_ONE", - seed=321, - ) - - client._build_url.assert_called_once_with("generate_video") - method, params = client._send.await_args.args[:2] - assert method == "api_request" - assert params["captchaAction"] == "VIDEO_GENERATION" - - body = params["body"] - assert body["useV2ModelConfig"] is True - assert set(body["mediaGenerationContext"]) == {"batchId"} - request = body["requests"][0] - assert request["videoModelKey"] == "abra_i2v_10s" - assert request["startImage"] == {"mediaId": "start-1"} - assert "endImage" not in request - assert request["seed"] == 321 - assert result["data"]["flowkitPolling"]["mode"] == "project_media" - assert result["data"]["flowkitPolling"]["project_id"] == "project-1" - - -@pytest.mark.asyncio -async def test_submit_builds_flow_omni_first_last_request(): - client = _mock_submit_client() - - with patch("agent.services.omni_flash.get_flow_client", return_value=client): - result = await generate_omni_flash_first_last_video( - start_image_media_id="start-1", - end_image_media_id="end-1", - prompt="Move naturally from the first composition to the final composition", - project_id="project-1", - scene_id="scene-1", - duration_s=8, - aspect_ratio="VIDEO_ASPECT_RATIO_PORTRAIT", - user_paygate_tier="PAYGATE_TIER_ONE", - seed=456, - ) - - client._build_url.assert_called_once_with("generate_video_start_end") - method, params = client._send.await_args.args[:2] - assert method == "api_request" - body = params["body"] - request = body["requests"][0] - assert request["videoModelKey"] == "abra_i2v_8s" - assert request["startImage"] == {"mediaId": "start-1"} - assert request["endImage"] == {"mediaId": "end-1"} - assert request["aspectRatio"] == "VIDEO_ASPECT_RATIO_PORTRAIT" - assert request["seed"] == 456 - assert result["data"]["flowkitPolling"]["workflows"] == [ - { - "name": "workflow-1", - "primary_media_id": "media-1", - "project_id": "project-1", - } - ] - +# ─── validation runs before the capability gap is reported ────────────────── +# +# Order matters: a caller with a bad duration AND an unported mode should hear +# about the duration, which it can fix, rather than the gap, which it cannot. @pytest.mark.asyncio async def test_first_frame_rejects_missing_start_before_submit(): with pytest.raises(ValueError, match="requires start_image_media_id"): await generate_omni_flash_first_frame_video( - start_image_media_id="", - prompt="test", - project_id="p", - ) + start_image_media_id="", prompt="test", project_id="p") @pytest.mark.asyncio async def test_first_last_rejects_missing_end_before_submit(): with pytest.raises(ValueError, match="non-empty end_image_media_id"): await generate_omni_flash_first_last_video( - start_image_media_id="start", - end_image_media_id="", - prompt="test", - project_id="p", - ) - - -@pytest.mark.asyncio -async def test_submit_builds_flow_omni_r2v_request_and_poll_descriptor(): - client = _mock_submit_client() - - with patch("agent.services.omni_flash.get_flow_client", return_value=client): - result = await generate_omni_flash_video( - reference_media_ids=["ref-1", "ref-2"], - prompt="Two friends talking in a cafe", - project_id="project-1", - scene_id="scene-1", - duration_s=10, - aspect_ratio="VIDEO_ASPECT_RATIO_LANDSCAPE", - user_paygate_tier="PAYGATE_TIER_ONE", - seed=123, - ) - - assert result["status"] == 200 - assert result["data"]["flowkitPolling"] == { - "mode": "project_media", - "project_id": "project-1", - "workflows": [ - { - "name": "workflow-1", - "primary_media_id": "media-1", - "project_id": "project-1", - } - ], - } - client._build_url.assert_called_once_with("generate_video_references") - client._send.assert_awaited_once() - - method, params = client._send.await_args.args[:2] - assert method == "api_request" - assert params["captchaAction"] == "VIDEO_GENERATION" - - body = params["body"] - assert body["useV2ModelConfig"] is True - assert body["mediaGenerationContext"]["audioFailurePreference"] == "BLOCK_SILENCED_VIDEOS" - assert body["clientContext"]["projectId"] == "project-1" - - request = body["requests"][0] - assert request["videoModelKey"] == "abra_r2v_10s" - assert request["aspectRatio"] == "VIDEO_ASPECT_RATIO_LANDSCAPE" - assert request["seed"] == 123 - assert request["metadata"] == {"sceneId": "scene-1"} - assert request["referenceImages"] == [ - {"mediaId": "ref-1", "imageUsageType": "IMAGE_USAGE_TYPE_ASSET"}, - {"mediaId": "ref-2", "imageUsageType": "IMAGE_USAGE_TYPE_ASSET"}, - ] - - -def _project_response(generation_status="MEDIA_GENERATION_STATUS_PENDING", include_media=True): - media = [] - if include_media: - media.append( - { - "name": "media-1", - "projectId": "project-1", - "workflowId": "workflow-1", - "mediaMetadata": { - "mediaStatus": {"mediaGenerationStatus": generation_status} - }, - "video": {"generatedVideo": {"model": "abra_r2v_4s"}}, - } - ) - return { - "status": 200, - "data": { - "result": { - "data": { - "json": { - "projectContents": { - "workflows": [ - { - "name": "workflow-1", - "projectId": "project-1", - "metadata": {"primaryMediaId": "media-1"}, - } - ], - "media": media, - } - } - } - } - }, - } - - -@pytest.mark.asyncio -async def test_project_poll_fetch_matches_live_flow_trpc_contract(): - client = MagicMock() - client._send = AsyncMock(return_value={"status": 200, "data": {}}) - - await _fetch_project_initial_data(client, "project-1") - - client._send.assert_awaited_once_with( - "trpc_request", - { - "url": ( - "https://labs.google/fx/api/trpc/flow.projectInitialData?input=" - "%7B%22json%22%3A%7B%22projectId%22%3A%22project-1%22%7D%7D" - ), - "method": "GET", - "headers": {"content-type": "application/json"}, - }, - timeout=15, - ) - - -@pytest.mark.asyncio -async def test_media_redirect_fetch_requests_url_only_mode(): - client = MagicMock() - client._send = AsyncMock(return_value={"status": 200, "data": {}}) - - await _fetch_media_url(client, "media-1") - - client._send.assert_awaited_once_with( - "trpc_request", - { - "url": ( - "https://labs.google/fx/api/trpc/media.getMediaUrlRedirect" - "?name=media-1" - ), - "method": "GET", - "headers": {"content-type": "application/json"}, - "responseMode": "url", - }, - timeout=15, - ) - - -@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() - client._send = AsyncMock(return_value=_project_response()) - client.get_media = AsyncMock(side_effect=AssertionError("legacy get_media forbidden")) - client.check_video_status = AsyncMock(side_effect=AssertionError("operation poll forbidden")) - - 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", - } - ] - ) - - client.get_media.assert_not_awaited() - client.check_video_status.assert_not_awaited() - client._send.assert_awaited_once() - assert result["done"] is False - assert result["status"] == "PENDING" - assert result["workflows"][0]["status"] == "PENDING" - - -@pytest.mark.asyncio -async def test_omni_poll_completed_returns_signed_url_without_buffering_video(): - client = MagicMock() - client._send = AsyncMock( - side_effect=[ - _project_response("MEDIA_GENERATION_STATUS_SUCCESSFUL"), - { - "status": 200, - "data": { - "url": "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["status"] == "COMPLETED" - item = result["workflows"][0] - assert item["status"] == "MEDIA_GENERATION_STATUS_SUCCESSFUL" - assert item["media"]["media_id"] == "media-1" - assert item["media"]["url"].startswith("https://flow-content.google/video/") - assert item["media"]["encoded_video_available"] is False - assert "encoded_video" not in item["media"] - - -@pytest.mark.asyncio -async def test_omni_poll_missing_media_is_pending(): - client = MagicMock() - client._send = AsyncMock(return_value=_project_response(include_media=False)) - - 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 False - assert result["workflows"][0]["status"] == "PENDING" - - -@pytest.mark.asyncio -async def test_omni_poll_requires_project_id_for_legacy_descriptors(): - with pytest.raises(ValueError, match="requires project_id"): - await check_omni_flash_status( - [{"name": "workflow-1", "primary_media_id": "media-1"}] - ) - - -@pytest.mark.asyncio -async def test_submit_accepts_seven_references(): - client = MagicMock() - client._client_context.return_value = {"projectId": "p"} - client._build_url.return_value = "https://example.test/omni" - client._send = AsyncMock(return_value={"status": 200, "data": {}}) - refs = [f"ref-{i}" for i in range(OMNI_FLASH_MAX_REFERENCE_IMAGES)] - - with patch("agent.services.omni_flash.get_flow_client", return_value=client): - await generate_omni_flash_video( - reference_media_ids=refs, - prompt="test", - project_id="p", - duration_s=4, - ) - - request = client._send.await_args.args[1]["body"]["requests"][0] - assert len(request["referenceImages"]) == 7 + start_image_media_id="start", end_image_media_id="", + prompt="test", project_id="p") @pytest.mark.asyncio async def test_submit_rejects_more_than_seven_references(): refs = [f"ref-{i}" for i in range(OMNI_FLASH_MAX_REFERENCE_IMAGES + 1)] - with pytest.raises(ValueError, match="at most 7 reference images"): await generate_omni_flash_video( - reference_media_ids=refs, - prompt="test", - project_id="p", - duration_s=8, - ) + reference_media_ids=refs, prompt="test", project_id="p", duration_s=8) @pytest.mark.asyncio async def test_submit_rejects_empty_reference_set(): with pytest.raises(ValueError, match="requires at least one reference image"): await generate_omni_flash_video( - reference_media_ids=[], - prompt="test", - project_id="p", - duration_s=8, - ) + reference_media_ids=[], prompt="test", project_id="p", duration_s=8) + + +@pytest.mark.asyncio +async def test_seven_references_pass_validation_and_reach_the_capability_gap(): + """Seven is the limit, not one past it — so this must fail on the gap, not + on the count. Guards the boundary against an off-by-one in the validator.""" + refs = [f"ref-{i}" for i in range(OMNI_FLASH_MAX_REFERENCE_IMAGES)] + result = await generate_omni_flash_video( + reference_media_ids=refs, prompt="test", project_id="p", duration_s=4) + assert "UNSUPPORTED_ON_BATCH_API" in result["error"] -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.""" +# ─── unported modes ───────────────────────────────────────────────────────── - @pytest.fixture(autouse=True) - def batch_transport(self, monkeypatch): - monkeypatch.setattr(omni_flash, "USE_BATCH_RPC", True) +class TestUnportedOmniModesAreRefusedRatherThanAttempted: + """Frame and reference generation had only a REST implementation, and that + transport is gone. Naming the gap beats a 401 five retries deep — and beats + silently submitting with a Veo key, which is what a careless unification + of these paths would do.""" @pytest.fixture def client(self): with patch("agent.services.omni_flash.get_flow_client") as factory: stub = MagicMock() stub._send = AsyncMock() + stub.generate_video = AsyncMock() factory.return_value = stub yield stub @@ -547,6 +205,7 @@ async def test_first_frame_names_the_gap_and_sends_nothing(self, client): start_image_media_id="mid", prompt="go", project_id="pid") assert "UNSUPPORTED_ON_BATCH_API" in result["error"] client._send.assert_not_called() + client.generate_video.assert_not_called() async def test_first_last_names_the_gap_and_sends_nothing(self, client): result = await generate_omni_flash_first_last_video( @@ -554,14 +213,18 @@ async def test_first_last_names_the_gap_and_sends_nothing(self, client): prompt="go", project_id="pid") assert "UNSUPPORTED_ON_BATCH_API" in result["error"] client._send.assert_not_called() + client.generate_video.assert_not_called() async def test_reference_to_video_names_the_gap_and_sends_nothing(self, client): result = await generate_omni_flash_video( reference_media_ids=["a"], prompt="go", project_id="pid") assert "UNSUPPORTED_ON_BATCH_API" in result["error"] client._send.assert_not_called() + client.generate_video.assert_not_called() async def test_the_message_points_to_supported_text_to_video(self, client): + """The error is the only place a caller learns there IS a working Omni + surface, so it names both the gap and the way round it.""" result = await generate_omni_flash_video( reference_media_ids=["a"], prompt="go", project_id="pid") assert "text-to-video is supported" in result["error"] diff --git a/tests/unit/test_operations.py b/tests/unit/test_operations.py index 670435164..c0b85e2da 100644 --- a/tests/unit/test_operations.py +++ b/tests/unit/test_operations.py @@ -159,6 +159,60 @@ async def test_uses_landscape_for_horizontal_orientation(self, service, base_sce assert call_kwargs["aspect_ratio"] == "IMAGE_ASPECT_RATIO_LANDSCAPE" +# --------------------------------------------------------------------------- +# Test: generate_scene_video — the retry branch +# --------------------------------------------------------------------------- + +class TestGenerateSceneVideoRetry: + """A retry must re-poll a submitted render, never resubmit it. + + This is the one branch where a wrong answer costs money: resubmitting + abandons a render Flow is already paying for and starts a second one. It + was previously gated on the transport (`bare_uuid and not USE_BATCH_RPC`), + which made it dead on the only transport that exists, so it went untested. + """ + + @pytest.mark.asyncio + async def test_a_submitted_operation_is_repolled_not_resubmitted( + self, service, base_scene, mock_client): + mock_client.generate_video = AsyncMock() + polled = {"data": {"operations": [{"status": "MEDIA_GENERATION_STATUS_SUCCESSFUL"}]}} + + with patch("agent.sdk.services.operations.crud") as mock_crud, \ + patch("agent.sdk.services.operations._build_video_prompt", + new=AsyncMock(return_value="p")), \ + patch("agent.sdk.services.operations._poll_operations", + new=AsyncMock(return_value=polled)) as mock_poll: + mock_crud.get_project = AsyncMock(return_value={"user_paygate_tier": "PAYGATE_TIER_ONE"}) + mock_crud.get_request = AsyncMock(return_value={"request_id": SAMPLE_UUID_2}) + + result = await service.generate_scene_video( + base_scene, "VERTICAL", request_id="req-1") + + mock_client.generate_video.assert_not_called() + assert mock_poll.await_args.args[1][0]["operation"]["name"] == SAMPLE_UUID_2 + assert result is polled + + @pytest.mark.asyncio + async def test_no_prior_operation_still_submits(self, service, base_scene, mock_client): + """The complement — the guard must not swallow a first attempt.""" + mock_client.generate_video = AsyncMock( + return_value={"data": {"operations": [{"operation": {"name": "op-new"}}]}}) + + with patch("agent.sdk.services.operations.crud") as mock_crud, \ + patch("agent.sdk.services.operations._build_video_prompt", + new=AsyncMock(return_value="p")), \ + patch("agent.sdk.services.operations._poll_operations", + new=AsyncMock(return_value={"data": {}})): + mock_crud.get_project = AsyncMock(return_value={"user_paygate_tier": "PAYGATE_TIER_ONE"}) + mock_crud.get_request = AsyncMock(return_value=None) + mock_crud.update_request = AsyncMock() + + await service.generate_scene_video(base_scene, "VERTICAL", request_id="req-1") + + mock_client.generate_video.assert_called_once() + + # --------------------------------------------------------------------------- # Test: edit_scene_image # --------------------------------------------------------------------------- diff --git a/tests/unit/test_processor.py b/tests/unit/test_processor.py index 2340a209c..c2f7e9850 100644 --- a/tests/unit/test_processor.py +++ b/tests/unit/test_processor.py @@ -139,8 +139,7 @@ async def test_retries_when_under_max_retries(self): rid = req["id"] result = {"error": "timeout"} - with patch("agent.worker.processor.crud") as mock_crud, \ - patch("agent.worker.processor._retry_state", {}): + with patch("agent.worker.processor.crud") as mock_crud: mock_crud.update_request = AsyncMock() mock_crud.update_scene = AsyncMock() await _handle_failure(rid, req, result) @@ -158,8 +157,7 @@ async def test_marks_failed_when_at_max_retries(self): rid = req["id"] result = {"error": "permanent failure"} - with patch("agent.worker.processor.crud") as mock_crud, \ - patch("agent.worker.processor._retry_state", {}): + with patch("agent.worker.processor.crud") as mock_crud: mock_crud.update_request = AsyncMock() mock_crud.update_scene = AsyncMock() await _handle_failure(rid, req, result) @@ -185,8 +183,7 @@ async def test_extracts_error_message_from_nested_data(self): } } - with patch("agent.worker.processor.crud") as mock_crud, \ - patch("agent.worker.processor._retry_state", {}): + with patch("agent.worker.processor.crud") as mock_crud: mock_crud.update_request = AsyncMock() mock_crud.update_scene = AsyncMock() await _handle_failure(rid, req, result) diff --git a/tests/unit/test_result_handler.py b/tests/unit/test_result_handler.py index 2641f64c7..6b893b38c 100644 --- a/tests/unit/test_result_handler.py +++ b/tests/unit/test_result_handler.py @@ -60,6 +60,10 @@ def test_parse_result_raw_is_attached(sample_image_success): @pytest.mark.asyncio async def test_apply_scene_result_generate_image_sets_fields_and_cascades(sample_uuid, mocker): mock_update = mocker.patch("agent.sdk.services.result_handler.crud.update_scene", new_callable=AsyncMock) + # The chain cascade reads the scene back; unmocked this opens a real aiosqlite + # connection, whose worker thread then outlives the test session. + mocker.patch("agent.sdk.services.result_handler.crud.get_scene", + new_callable=AsyncMock, return_value=None) result = GenerationResult(success=True, media_id=sample_uuid, url="https://example.com/img.jpg") await apply_scene_result("scene-001", "GENERATE_IMAGE", "VERTICAL", result) @@ -78,6 +82,10 @@ async def test_apply_scene_result_generate_image_sets_fields_and_cascades(sample @pytest.mark.asyncio async def test_apply_scene_result_edit_image_same_cascade_as_generate(sample_uuid, mocker): mock_update = mocker.patch("agent.sdk.services.result_handler.crud.update_scene", new_callable=AsyncMock) + # The chain cascade reads the scene back; unmocked this opens a real aiosqlite + # connection, whose worker thread then outlives the test session. + mocker.patch("agent.sdk.services.result_handler.crud.get_scene", + new_callable=AsyncMock, return_value=None) result = GenerationResult(success=True, media_id=sample_uuid, url="https://example.com/img.jpg") await apply_scene_result("scene-001", "EDIT_IMAGE", "VERTICAL", result) @@ -146,6 +154,10 @@ async def test_apply_scene_result_skips_when_result_failed(mocker): @pytest.mark.asyncio async def test_apply_scene_result_horizontal_orientation_uses_correct_prefix(sample_uuid, mocker): mock_update = mocker.patch("agent.sdk.services.result_handler.crud.update_scene", new_callable=AsyncMock) + # The chain cascade reads the scene back; unmocked this opens a real aiosqlite + # connection, whose worker thread then outlives the test session. + mocker.patch("agent.sdk.services.result_handler.crud.get_scene", + new_callable=AsyncMock, return_value=None) result = GenerationResult(success=True, media_id=sample_uuid, url="https://example.com/img.jpg") await apply_scene_result("scene-001", "GENERATE_IMAGE", "HORIZONTAL", result)