"""Flora provider adapter — wraps Flora's POST /generate API. Flora charges per-second (~$0.21/s standard_720p, PROVISIONAL pending first invoice). The adapter treats Flora as a pure execution transport; provenance is captured at dispatch time in the Recoil manifest, not read back from Flora (Flora's API is write-forward). """ from __future__ import annotations import base64 import hashlib import json import os import tempfile from datetime import datetime, timezone from pathlib import Path from typing import Optional from recoil.execution.providers.flora_projects import ( _cache_read, _cache_update, resolve_flora_project, ) from recoil.execution.providers.base import ( CAPABILITY_KEYS, PollRequest, PollResult, ProviderCapabilityError, ProviderJob, SubmitRequest, UnifiedVideoPayload, ) _BASE_URL = "https://app.flora.ai/api/v1" _FLORA_SUPPORTED_RESOLUTIONS = {"720p"} # Flora uses modality-prefixed model IDs. The tuple key is # (recoil_model_id, action) where action comes from _infer_action(). _FLORA_MODEL_IDS = { ("seeddance-2.0", "t2v"): "t2v-seedance-2.0-enhancor", ("seeddance-2.0", "i2v"): "i2v-seedance-2.0-enhancor", ("seeddance-2.0", "r2v"): "r2v-seedance-2.0-enhancor", ("seeddance-2.0", "f2v"): "f2v-seedance-2.0-enhancor", ("seeddance-2.0", "v2v"): "v2v-seedance-2.0-reference-enhancor", ("seeddance-2.0-fast", "t2v"): "t2v-seedance-2.0-fast-enhancor", ("seeddance-2.0-fast", "i2v"): "i2v-seedance-2-fast-enhancor", ("seeddance-2.0-fast", "r2v"): "r2v-seedance-2.0-fast-enhancor", ("seeddance-2.0-fast", "f2v"): "f2v-seedance-2-fast-enhancor", ("seeddance-1.5-pro", "t2v"): "t2v-seedance-1.5-pro", ("seeddance-1.5-pro", "i2v"): "i2v-seedance-1.5-pro", ("seeddance-1.5-pro", "f2v"): "f2v-seedance-pro", # ---- Kling video (LIVE GET /models 2026-06-23; see consultations/recoil/ # flora-kling-adapter-2026-06-23/LIVE_KLING_ENDPOINTS.md) ---- # kling-o3-pro = the "omni" flagship; full action set incl. r2v (refs→video). ("kling-o3-pro", "t2v"): "t2v-kling-o3", ("kling-o3-pro", "i2v"): "i2v-kling-o3", ("kling-o3-pro", "f2v"): "f2v-kling-o3", ("kling-o3-pro", "r2v"): "r2v-kling-o3", # kling-3.0-pro = Kling v3. Plain `i2v-kling-v3-pro` is a DEAD endpoint that # does NOT exist live (would 404) — the only live 3.0-Pro i2v route is the # Pro-Turbo variant `i2v-kling-3-0-pro-turbo-i2v`. v3 has NO r2v endpoint # (r2v is omni-only on kling-o3-pro). ("kling-3.0-pro", "t2v"): "t2v-kling-v3-pro", ("kling-3.0-pro", "i2v"): "i2v-kling-3-0-pro-turbo-i2v", ("kling-3.0-pro", "f2v"): "f2v-kling-v3-pro", ("kling-2.5-pro", "t2v"): "t2v-kling-2.5", ("kling-2.5-pro", "i2v"): "i2v-kling-2.5", ("kling-2.5-pro", "f2v"): "f2v-kling-2.5-pro", ("gpt-image-2", "t2i"): "t2i-gpt-image-2", # ---- Image models (Phase 0, krea2-flora-image-pipeline) ---- # Verified live in Flora GET /models?type=image. Do NOT add a model_id # that isn't in the verified live roster: Flora silently falls back to a # default model AND BILLS on an unknown model_id (footgun, verified). # is2i = identity/style-reference image gen; t2i = text-to-image concepting. ("krea-2", "t2i"): "t2i-krea-2-t2i", ("krea-2-references", "is2i"): "is2i-krea-2-references-is2i", ("seedream-v4.5", "t2i"): "t2i-seedream-v4.5", ("seedream-v4.5", "is2i"): "is2i-seedream-v4.5", ("seedream-v5", "is2i"): "is2i-seedream-v5", ("nano-banana", "t2i"): "t2i-nano-banana", ("nano-banana", "is2i"): "is2i-nano-banana", ("gemini-3-pro", "t2i"): "t2i-gemini-3-pro", ("gpt-image-2", "is2i"): "is2i-gpt-image-2", ("gemini-3.1-flash-image-preview", "is2i"): "is2i-gemini-3.1-flash-image", } _SUPPORTED_MODELS = sorted({k[0] for k in _FLORA_MODEL_IDS}) # Image actions Flora exposes. t2i = text-to-image; is2i = identity/style # reference-conditioned image gen; i2i = image-to-image edit. A recoil model # that has ANY of these actions registered is treated as an image model. _IMAGE_ACTIONS = frozenset({"t2i", "i2i", "is2i"}) _IMAGE_MODELS = frozenset( k[0] for k in _FLORA_MODEL_IDS if k[1] in _IMAGE_ACTIONS ) # Image models that take reference images (is2i / i2i variants registered). _IMAGE_REF_MODELS = frozenset( k[0] for k in _FLORA_MODEL_IDS if k[1] in ("is2i", "i2i") ) # SINGLE POINT OF TRUTH for the Flora image reference-input param name. # CONFIRMED 2026-06-23 (REC-246) via a live i2i edit-pass: a source frame placed # at params["image_urls"] produced a composition-faithful recolor (red cube → # blue, same framing/shadow/background) — so `image_urls` (a list, inside # `params`) IS the correct image-input field. Both the i2i source frame # (payload.image) and is2i identity refs (payload.reference_images) are appended # to this one list in _build_params, so the placement covers both. (The # "params are tuning-only" caveat from GET /models applies to VIDEO models; image # models accept image_urls in params.) If a future model needs a different field, # change THIS constant only. _IMAGE_REF_PARAM = "image_urls" # SINGLE POINT OF TRUTH for where the VIDEO image-INPUT fields go in the # POST /generate body. CONFIRMED 2026-06-23 via a live seedance-2.0 i2v probe: # a start frame at params["image_url"] conditioned the output — the video's # frame 0 was pixel-identical to the start frame — so `params` IS the correct # placement for the VIDEO image inputs (i2v start `image_url`, and by the same # mechanism f2v tail `end_image_url`, r2v refs `image_urls`, r2v `video_urls`). # The "params are tuning-only" note from GET /models did NOT mean the image # inputs must be top-level. Leave False; flip this ONE constant only if a future # model ever rejects refs in `params`. # False = current/confirmed: image-input fields stay inside `params`. # True = image-input fields move to the top level of the POST body. # Tuning keys (aspect_ratio / duration / resolution / seed / generate_audio / # negative_prompt) ALWAYS stay in `params` regardless of this flag. _FLORA_VIDEO_IMAGE_FIELDS_AT_TOPLEVEL: bool = False # The VIDEO image-INPUT keys relocated by _FLORA_VIDEO_IMAGE_FIELDS_AT_TOPLEVEL. # Everything else _build_params emits is a tuning key and stays in `params`. _FLORA_VIDEO_IMAGE_INPUT_KEYS = ( "image_url", "end_image_url", "image_urls", "video_urls", ) # Live dynamic-endpoints probe 2026-06-09: every Seedance video route's # preprocessing sets maxPromptLength=2500 with blankPrompt='. _ .' — # over-length prompts are blanked, not truncated (REC-123). _FLORA_MAX_PROMPT_CHARS = 2500 def _file_sha256(path: str | Path) -> str: h = hashlib.sha256() with Path(path).open("rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def _cache_path(project: str) -> Path: from recoil.core.paths import projects_root return ( projects_root() / project / "_pipeline" / "state" / "flora" / "upload_cache.json" ) def _infer_action(payload: UnifiedVideoPayload) -> str: if payload.model_id in _IMAGE_MODELS: # Image path: a ref image (start frame) or reference_images promotes # the action to i2i/is2i, mirroring how video promotes t2v→i2v/r2v. # Prefer is2i when the model registers it (identity/style strict); # fall back to i2i for plain image-to-image edit models. if payload.reference_images or payload.image: if (payload.model_id, "is2i") in _FLORA_MODEL_IDS: return "is2i" if (payload.model_id, "i2i") in _FLORA_MODEL_IDS: return "i2i" return "t2i" if payload.reference_images or payload.reference_videos: return "r2v" if payload.image and payload.image_tail: return "f2v" if payload.image: return "i2v" return "t2v" def _build_params(payload: UnifiedVideoPayload, action: str) -> dict: params: dict = {} # ---- Image actions (t2i / i2i / is2i) ---------------------------------- # Image-model params are tuning-only per live GET /models # (creativity / seed / aspect_ratio / resolution). Duration / audio / # video-resolution gating below are VIDEO concepts and do not apply, so the # image branch returns early. if action in _IMAGE_ACTIONS: if payload.aspect_ratio: params["aspect_ratio"] = payload.aspect_ratio # Image-model resolution is a free-form tuning enum (e.g. "2K"), NOT # the per-second video resolution gate. UnifiedVideoPayload.resolution # DEFAULTS to the video token "720p" (the sole member of # _FLORA_SUPPORTED_RESOLUTIONS), which is never a valid image-model # resolution — so skip it here and only forward an explicit image # resolution the caller actually set. Do not validate against the video # set; an explicit image value passes through verbatim. if payload.resolution and payload.resolution not in _FLORA_SUPPORTED_RESOLUTIONS: params["resolution"] = payload.resolution # creativity / seed arrive via payload.hints (Krea-2 creativity dial, # seed) and are merged through coerce_to_dict() at the end of this fn. if action in ("i2i", "is2i"): # Flora image ref-input field — CONFIRMED 2026-06-23 (REC-246). # --------------------------------------------------------------- # A live i2i edit-pass (nano-banana, source frame → params[ # "image_urls"]=[hosted_url]) produced a composition-faithful recolor, # proving `image_urls` (a list, inside `params`) IS the correct # image-input field — the "params are tuning-only" note from GET # /models applies to VIDEO models, not image. Both the i2i source # frame and is2i identity refs share this one list. _IMAGE_REF_PARAM # (module top) remains the single point of truth if a future model # ever needs a different field. ref_urls: list = [] if payload.reference_images: ref_urls.extend(payload.reference_images) if payload.image and payload.image not in ref_urls: ref_urls.append(payload.image) if ref_urls: params[_IMAGE_REF_PARAM] = ref_urls if payload.hints: from recoil.execution.providers.payload_hints import coerce_to_dict # WHITELIST the hint keys forwarded to Flora's image params. For # image keyframes, StepRunnerHints (extra="allow") always carries # the Google/fal-shape keys modality/parts/genai_config (+ quality, # size_override, refs_used, start_frame, elements_payload). Flora's # image /generate params are tuning-only (creativity / seed / # aspect_ratio / resolution) — dumping the whole hints dict would # leak unknown keys into the body (silent-default-model billing # footgun). Only forward keys Flora actually accepts. Extend this set # as more image params are confirmed via the supervised probe. _h = coerce_to_dict(payload.hints) for _k in ("creativity", "seed"): if _h.get(_k) is not None: params[_k] = _h[_k] return params if payload.duration_s: params["duration"] = str(int(payload.duration_s)) if payload.resolution: if payload.resolution not in _FLORA_SUPPORTED_RESOLUTIONS: raise ValueError( f"Flora does not support resolution {payload.resolution!r}. " f"Supported: {_FLORA_SUPPORTED_RESOLUTIONS}" ) params["resolution"] = payload.resolution if payload.aspect_ratio: params["aspect_ratio"] = payload.aspect_ratio if payload.generate_audio is not None: params["generate_audio"] = payload.generate_audio if action == "i2v" and payload.image: params["image_url"] = payload.image elif action == "f2v": if payload.image: params["image_url"] = payload.image if payload.image_tail: params["end_image_url"] = payload.image_tail elif action == "r2v": if payload.reference_images: params["image_urls"] = payload.reference_images if payload.reference_videos: params["video_urls"] = payload.reference_videos if payload.image: params["image_url"] = payload.image if payload.negative_prompt: params["negative_prompt"] = payload.negative_prompt if payload.hints: from recoil.execution.providers.payload_hints import coerce_to_dict params.update(coerce_to_dict(payload.hints)) return params def _flora_host_image_value( value: str, api_key: str, workspace_id: str, project_id: str, project: Optional[str] = None, ) -> str: """Normalize ONE image input to a Flora-hosted https URL. Flora's /generate needs hosted URLs, but the canonical UnifiedVideoPayload may carry a frame as an http(s) URL, a local file path, OR base64 image bytes (the inline shape the fal path uses; dispatch_payload base64-encodes i2v/f2v frames). The Flora layer normalizes all three here: - http(s) URL -> returned unchanged - existing local file -> uploaded directly - base64 / data: URI -> decoded to a temp file, then uploaded """ if value.startswith(("http://", "https://")): return value # An existing local file path uploads directly. Guard the exists() probe # against OSError — a multi-MB base64 string is not a valid filename. try: is_file = len(value) < 4096 and Path(value).is_file() except OSError: is_file = False if not is_file: # Treat as base64 image bytes (optionally a data: URI). b64 = value.split(",", 1)[-1] if value.startswith("data:") else value try: raw = base64.b64decode(b64, validate=False) except Exception as e: # noqa: BLE001 — any decode failure = bad input raise RuntimeError( "Flora image input is neither an http(s) URL, an existing file, " f"nor decodable base64 ({e})" ) from e with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tf: tf.write(raw) value = tf.name return _upload_local_refs( [value], api_key, workspace_id, project_id, project=project )[0] def _upload_local_refs( local_paths: list[str], api_key: str, workspace_id: str, project_id: str, project: Optional[str] = None, ) -> list[str]: """Upload local image files to Flora and return hosted asset URLs. Flora's /generate endpoint requires hosted URLs for reference images. Flow per file: 1. POST /assets (source=signed-url) → signed upload URL + asset_id 2. PUT the file bytes to the signed URL 3. POST /assets/{id}/complete 4. POST /projects/{id}/assets/{id}/attach → canvas node 5. Return the asset's hosted URL Skips paths that are already URLs (http/https). """ import mimetypes import urllib.error import urllib.request headers_base = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", "User-Agent": "recoil-pipeline/1.0", } base = _BASE_URL hosted_urls: list[str] = [] cache_file = _cache_path(project) if project else None def _attach_asset(asset_id: str) -> None: attach_req = urllib.request.Request( f"{base}/projects/{project_id}/assets/{asset_id}/attach", headers=headers_base, method="POST", ) try: with urllib.request.urlopen(attach_req, timeout=30) as resp: resp.read() except urllib.error.HTTPError as e: err = e.read().decode() if e.fp else "" raise RuntimeError( f"Flora asset attach failed: HTTP {e.code}: {err}" ) from e def _drop_cache_entry(cache_key: str) -> None: if cache_file is None: return def _mutate(data: dict) -> dict: data.pop(cache_key, None) return data _cache_update(cache_file, _mutate) for path_str in local_paths: if path_str.startswith("http://") or path_str.startswith("https://"): hosted_urls.append(path_str) continue p = Path(path_str) if not p.exists(): raise FileNotFoundError(f"Flora ref upload: file not found: {p}") ct = mimetypes.guess_type(str(p))[0] or "image/png" cache_key = _file_sha256(p) if cache_file is not None else None if cache_file is not None and cache_key is not None: entry = _cache_read(cache_file).get(cache_key) if ( isinstance(entry, dict) and entry.get("project_id") == project_id and entry.get("asset_id") and entry.get("hosted_url") ): try: _attach_asset(str(entry["asset_id"])) hosted_urls.append(str(entry["hosted_url"])) continue except RuntimeError as e: # Flora 404s cached asset IDs as plain RuntimeError text # today; brittle string match until the transport exposes # typed status codes at this layer. if "HTTP 404" in str(e): _drop_cache_entry(cache_key) else: raise # Step 1: reserve signed upload reserve_body = json.dumps( { "source": "signed-url", "workspace_id": workspace_id, "content_type": ct, "file_name": p.name, "folder": "recoil-refs", } ).encode() req = urllib.request.Request( f"{base}/assets", data=reserve_body, headers=headers_base, method="POST", ) try: with urllib.request.urlopen(req, timeout=30) as resp: reserve = json.loads(resp.read().decode()) except urllib.error.HTTPError as e: err = e.read().decode() if e.fp else "" raise RuntimeError( f"Flora asset reserve failed: HTTP {e.code}: {err}" ) from e asset_id = reserve.get("asset_id", "") hosted_url = reserve.get("url", "") upload_info = reserve.get("upload", {}) upload_endpoint = upload_info.get("url", "") upload_method = upload_info.get("method", "POST").upper() file_field = upload_info.get("file_field", "file") form_fields = upload_info.get("form_fields", {}) if not upload_endpoint or not asset_id: raise RuntimeError(f"Flora asset reserve missing upload info: {reserve}") # Step 2: upload file via ImageKit multipart form file_bytes = p.read_bytes() boundary = "----RecoilUploadBoundary" body_parts: list[bytes] = [] for k, v in form_fields.items(): body_parts.append( f'--{boundary}\r\nContent-Disposition: form-data; name="{k}"\r\n\r\n{v}\r\n'.encode() ) body_parts.append( f'--{boundary}\r\nContent-Disposition: form-data; name="{file_field}"; filename="{p.name}"\r\nContent-Type: {ct}\r\n\r\n'.encode() + file_bytes + f"\r\n--{boundary}--\r\n".encode() ) multipart_body = b"".join(body_parts) upload_req = urllib.request.Request( upload_endpoint, data=multipart_body, headers={ "Content-Type": f"multipart/form-data; boundary={boundary}", "User-Agent": "recoil-pipeline/1.0", }, method=upload_method, ) try: with urllib.request.urlopen(upload_req, timeout=120) as resp: resp.read() except urllib.error.HTTPError as e: err = e.read().decode() if e.fp else "" raise RuntimeError(f"Flora file upload failed: HTTP {e.code}: {err}") from e # Step 3: complete complete_req = urllib.request.Request( f"{base}/assets/{asset_id}/complete", headers=headers_base, method="POST", ) try: with urllib.request.urlopen(complete_req, timeout=30) as resp: complete_resp = json.loads(resp.read().decode()) except urllib.error.HTTPError as e: err = e.read().decode() if e.fp else "" raise RuntimeError( f"Flora asset complete failed: HTTP {e.code}: {err}" ) from e # Step 4: attach to project canvas _attach_asset(asset_id) final_url = ( complete_resp.get("url") or complete_resp.get("download_url") or hosted_url or f"asset://{asset_id}" ) if cache_file is not None and cache_key is not None: cache_entry = { "asset_id": asset_id, "project_id": project_id, "uploaded_at": datetime.now(timezone.utc).isoformat(), "hosted_url": final_url, } def _mutate(data: dict) -> dict: data[cache_key] = cache_entry return data _cache_update(cache_file, _mutate) hosted_urls.append(final_url) return hosted_urls class FloraAdapter: provider_id = "flora" supported_models = _SUPPORTED_MODELS auth_env_var = "FLORA_API_KEY" base_url = _BASE_URL max_prompt_chars = _FLORA_MAX_PROMPT_CHARS capabilities = {k: False for k in CAPABILITY_KEYS} capabilities.update( { "t2v": True, "i2v": True, "r2v": True, "end_frame": True, "audio": True, "negative_prompt": True, "resolution_720p": True, } ) status = "primary" def build_submit(self, payload: UnifiedVideoPayload, tier: str) -> SubmitRequest: api_key = os.environ.get(self.auth_env_var) if not api_key: raise RuntimeError(f"{self.auth_env_var} environment variable not set.") # Flora's video-model preprocessing enforces maxPromptLength=2500 # and REPLACES an over-length prompt with the blank sentinel # instead of truncating — the run completes and bills with no # prompt at all (REC-123, EP001 2026-06-09). Fail loud here so a # too-long prompt can never silently burn a billed run. if ( payload.model_id not in _IMAGE_MODELS and payload.prompt and len(payload.prompt) > _FLORA_MAX_PROMPT_CHARS ): raise ValueError( f"Flora video prompt is {len(payload.prompt)} chars — over the " f"{_FLORA_MAX_PROMPT_CHARS}-char preprocessing cap; Flora would " "silently blank it (REC-123). Budget the prompt before dispatch." ) workspace_id = os.environ.get("RECOIL_FLORA_WORKSPACE") if not workspace_id: raise RuntimeError( "RECOIL_FLORA_WORKSPACE environment variable not set. " "Flora requires a workspace_id (starts with ws_)." ) project_hint = None episode_hint = None if payload.hints: from recoil.execution.providers.payload_hints import coerce_to_dict hints = coerce_to_dict(payload.hints) project_hint = hints.get("project") episode_hint = hints.get("episode") if project_hint and episode_hint is not None: project_id = resolve_flora_project( str(project_hint), int(episode_hint), api_key=api_key, workspace_id=workspace_id, create=True, ) else: project_id = os.environ.get("RECOIL_FLORA_PROJECT") if not project_id: raise RuntimeError( "RECOIL_FLORA_PROJECT environment variable not set. " "Flora requires a project_id (starts with prj_)." ) # Normalize every image INPUT to a Flora-hosted https URL before # building the request. _infer_action checks reference_images to decide # r2v vs t2v, and _build_params emits image/image_tail into the # image_url/end_image_url fields, so all of these must be hosted first. # _flora_host_image_value accepts http(s) URLs (no-op), local file # paths, AND base64 frame bytes (the inline shape dispatch_payload # produces for i2v/f2v), decoding base64 to a temp file before upload. _host = lambda v: _flora_host_image_value( # noqa: E731 v, api_key, workspace_id, project_id, project=project_hint ) if payload.reference_images: payload = UnifiedVideoPayload( **{ **payload.__dict__, "reference_images": [_host(r) for r in payload.reference_images], } ) # i2v start frame / i2i/is2i source frame. if isinstance(payload.image, str) and payload.image: payload = UnifiedVideoPayload( **{**payload.__dict__, "image": _host(payload.image)} ) # f2v end/tail frame. Without hosting it Flora would receive an # unreadable local path / raw base64 and silently drop the end frame # (the exact start+end drop class that shelved the protocol). if isinstance(payload.image_tail, str) and payload.image_tail: payload = UnifiedVideoPayload( **{**payload.__dict__, "image_tail": _host(payload.image_tail)} ) action = _infer_action(payload) flora_model = _FLORA_MODEL_IDS.get((payload.model_id, action)) if not flora_model: raise ProviderCapabilityError( model_id=payload.model_id, provider_id=self.provider_id, capability=action, supported_providers=[ k[1] for k in _FLORA_MODEL_IDS if k[0] == payload.model_id ], ) params = _build_params(payload, action) # Image actions (t2i / i2i / is2i) → "image"; everything else is video. # Keyed off the resolved action (not a flora_model prefix) so is2i / i2i # model ids classify correctly. gen_type = "image" if action in _IMAGE_ACTIONS else "video" # VIDEO image-INPUT field placement (single point of truth above). When # _FLORA_VIDEO_IMAGE_FIELDS_AT_TOPLEVEL is True, lift the image-input # keys out of `params` (where _build_params puts them) to the body top # level. Tuning keys stay in `params`. No-op for image actions. toplevel_image_fields: dict = {} if ( action not in _IMAGE_ACTIONS and _FLORA_VIDEO_IMAGE_FIELDS_AT_TOPLEVEL ): for _k in _FLORA_VIDEO_IMAGE_INPUT_KEYS: if _k in params: toplevel_image_fields[_k] = params.pop(_k) body: dict = { "prompt": payload.prompt, "type": gen_type, "model": flora_model, "params": params, "workspace_id": workspace_id, "project_id": project_id, } body.update(toplevel_image_fields) return SubmitRequest( method="POST", url=f"{self.base_url}/generate", headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", # Vercel WAF returns a fake HTTP 401 "Invalid API key" when the # UA is missing/urllib (see feedback-vercel-waf-blocks-python-urllib-ua). "User-Agent": "recoil-pipeline/1.0", }, body=body, ) def parse_submit( self, resp: dict, payload: UnifiedVideoPayload, tier: str ) -> ProviderJob: run_id = resp.get("run_id") if not run_id: raise RuntimeError(f"Flora submit response missing run_id: {resp}") poll_url = resp.get("poll_url", f"{self.base_url}/runs/{run_id}") return ProviderJob( provider_id=self.provider_id, model_id=payload.model_id, native_id=run_id, tier=tier, duration_s=payload.duration_s, resolution=payload.resolution, native_state={ "poll_url": poll_url, "charged_cost": resp.get("charged_cost", 0.0), "estimated_seconds": resp.get("estimated_seconds"), }, ) def build_poll(self, job: ProviderJob) -> PollRequest: api_key = os.environ.get(self.auth_env_var) if not api_key: raise RuntimeError(f"{self.auth_env_var} environment variable not set.") poll_url = job.native_state.get( "poll_url", f"{self.base_url}/runs/{job.native_id}" ) return PollRequest( method="GET", url=poll_url, headers={ "Authorization": f"Bearer {api_key}", "User-Agent": "recoil-pipeline/1.0", # Vercel WAF — see submit_request }, ) def parse_poll(self, resp: dict, job: ProviderJob) -> PollResult: status_raw = resp.get("status", "pending") status_map = { "pending": "IN_PROGRESS", "running": "IN_PROGRESS", "completed": "COMPLETED", "failed": "FAILED", } status = status_map.get(status_raw, "FAILED") # PollResult.video_url is the generic output-URL slot for this adapter # (image runs reuse it). Match video AND image output types so image # gens (t2i/i2i/is2i) return their hosted URL like video does. video_url = None if status == "COMPLETED": outputs = resp.get("outputs", []) for out in outputs: if out.get("type") in ("videoUrl", "video", "imageUrl", "image"): video_url = out.get("url") break # Flora failures carry error_code (e.g. BILLING_NOT_ENOUGH_CREDITS), # not error_message. Surface whichever is present so take records # show the real provider reason instead of a generic # "provider reported FAILED" (REC-122). error = resp.get("error_message") error_code = resp.get("error_code") if error_code: error = f"{error} [{error_code}]" if error else f"flora: {error_code}" return PollResult( status=status, video_url=video_url, audio_url=None, observed_cost=job.native_state.get("charged_cost", 0.0), error=error, raw=resp, ) def build_result_fetch(self, job: ProviderJob) -> Optional[PollRequest]: return None def parse_result(self, resp: dict, job: ProviderJob) -> PollResult: return self.parse_poll(resp, job) def _http_json(self, method: str, url: str, headers: dict, body) -> dict: """POST/GET JSON against Flora, returning the decoded response dict. The single HTTP seam for direct_submit_image — tests monkeypatch this to avoid the network. Matches the urllib + mandatory UA pattern used by _flora_host_image_value (the Vercel WAF 401s a missing/urllib UA). """ import json as _json import urllib.error import urllib.request data = _json.dumps(body).encode() if body is not None else None req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=120) as resp: return _json.loads(resp.read().decode()) except urllib.error.HTTPError as e: detail = e.read().decode(errors="replace")[:500] raise RuntimeError(f"Flora HTTP {e.code} on {method} {url}: {detail}") from e def direct_submit_image(self, payload: UnifiedVideoPayload) -> dict: """Synchronous image dispatch — wraps Flora's async submit/poll/download. ``execute_keyframe()`` dispatches image generation through this synchronous-bypass method (``google.py`` / ``fal.py`` implement it for their providers). Flora is async (POST /generate → poll GET /runs/{id}), so this wraps that loop: ``build_submit`` (which hosts every image input AND only ever emits a VALIDATED ``flora_model`` — an unknown id raises ProviderCapabilityError rather than silently falling back to a default model and billing) → POST → poll ``parse_poll`` until COMPLETED → download the hosted output. Returns ``{"image_bytes", "cost_usd", "native_id"}`` for ``execute_keyframe`` (which reads image_bytes + cost_usd at step_runner.py:1952/1958). The image ref-input field placement (``_IMAGE_REF_PARAM``) for is2i/i2i remains the one empirically-unconfirmed bit — t2i (no ref) is fully exercised by this path; an is2i run needs a visual confirmation that the ref took. """ import time as _time from recoil.execution.lib.http_helpers import _download_video # Pre-submit cost cap (mirror FalAdapter.direct_submit_image): refuse # BEFORE billing if the per-shot estimate exceeds the profile cap. Best # effort — skip silently when no cap or no cost is known (get_cost may # raise CostMissingError for a cost-less model). try: from recoil.core import model_profiles _cap = (model_profiles.get_profile(payload.model_id) or {}).get( "max_cost_per_shot_usd" ) if _cap is not None: try: _est = model_profiles.get_cost(payload.model_id) except Exception: # noqa: BLE001 — cost optional for the cap check _est = None if _est is not None and _est > float(_cap): raise ValueError( f"Flora image cost estimate ${_est:.3f} for " f"{payload.model_id} exceeds cap ${float(_cap):.3f}" ) except ValueError: raise except Exception: # noqa: BLE001 — cap check is best-effort pass tier = "default" submit_req = self.build_submit(payload, tier) submit_resp = self._http_json( submit_req.method, submit_req.url, submit_req.headers, submit_req.body ) job = self.parse_submit(submit_resp, payload, tier) timeout_s = 300 deadline = _time.monotonic() + timeout_s interval = 3.0 while True: poll_req = self.build_poll(job) poll_resp = self._http_json( poll_req.method, poll_req.url, poll_req.headers, None ) result = self.parse_poll(poll_resp, job) if result.status == "COMPLETED": if not result.video_url: raise RuntimeError( f"Flora image run {job.native_id} COMPLETED with no output URL" ) image_bytes = _download_video(result.video_url) if not image_bytes: raise RuntimeError( f"Flora image download returned no bytes ({result.video_url})" ) return { "image_bytes": image_bytes, "cost_usd": float(result.observed_cost or 0.0), "native_id": job.native_id, } if result.status == "FAILED": raise RuntimeError(f"Flora image generation failed: {result.error}") if _time.monotonic() >= deadline: raise TimeoutError( f"Flora image run {job.native_id} did not complete within {timeout_s}s" ) _time.sleep(interval) interval = min(interval * 1.5, 15.0) def compute_cost(self, duration_s: float, tier: str, profile: dict) -> float: flora_profile = ( profile.get("providers", {}).get("flora", {}).get("tiers", {}).get(tier, {}) ) flat_rate = flora_profile.get("cost_per_run") if flat_rate is not None: return float(flat_rate) per_sec = flora_profile.get("cost_per_second", 0.0) return float(per_sec) * float(duration_s) ADAPTER = FloraAdapter()