"""Fal.ai adapter for seeddance-2.0. Extracted from SeedDanceClient (recoil/execution/api_client.py) in the Provider Adapter Refactor (2026-04-17). Preserves exact wire behavior: - POST queue endpoint, receive request_id + status_url + response_url - GET status_url, GET response_url on COMPLETED - image (base64) -> temp file -> fal_client.upload_file -> image_url - image_tail (base64) -> end_image_url - reference_images (local paths or http URLs) -> image_urls list Tier -> endpoint map: standard_720p -> bytedance/seedance-2.0/{action} (no sub-path) fast_720p -> bytedance/seedance-2.0/fast/{action} fast_480p -> bytedance/seedance-2.0/fast/{action}, resolution=480p """ from __future__ import annotations import base64 import logging import os import time import urllib.error import urllib.request from typing import Optional from recoil.execution.providers.base import ( PollRequest, PollResult, ProviderJob, SubmitRequest, UnifiedVideoPayload, ) from recoil.execution.providers.fal_transport import ( QUEUE_BASE as _QUEUE_BASE, upload_bytes_to_fal as _upload_bytes_to_fal, ) from recoil.execution.providers.payload_hints import coerce_to_dict from recoil.execution.providers import fal_transport as _fal_transport logger = logging.getLogger(__name__) # Re-exported for backwards compat with anything importing _fal_client / # _HAS_FAL from this module. Sourced from fal_transport so behavior stays # byte-identical at the wire level. _fal_client = _fal_transport._fal_client _HAS_FAL = _fal_transport._HAS_FAL _ACTION_MAP = { "t2v": "text-to-video", "i2v": "image-to-video", "r2v": "reference-to-video", } # --------------------------------------------------------------- # gpt-image-2 module-level helpers # --------------------------------------------------------------- # gpt-image-2 endpoints (SYNTHESIS §3). One adapter, payload-discriminated. _GPT_IMAGE_2_ACTION_MAP = { "gpt_image_2_t2i": "fal-ai/gpt-image-2", # Edit endpoint: the 2026-05-19 wire build hardcoded # "fal-ai/gpt-image-2/image-to-image" which returns 404 on the live API # (no such path). Corrected 2026-05-25 to "fal-ai/gpt-image-2/edit" per # the live fal.ai docs page (fal.ai/models/openai/gpt-image-2/edit). # The mock test asserted against the wrong value, so the breakage went # undetected until the [:1] cap fix in this same session finally let # a multi-ref call escape to the live endpoint. "gpt_image_2_edit": "fal-ai/gpt-image-2/edit", } # Aspect-ratio → pixel-dimension translation (SYNTHESIS §11). _GPT_IMAGE_2_SIZE_MAP = { "1:1": "1024x1024", "9:16": "1024x1536", "16:9": "1536x1024", } _GPT_IMAGE_2_DEFAULT_QUALITY = "medium" # Custom-dimension API constraints (OpenAI public docs 2026-05; confirmed JT # 2026-05-26). When a caller passes a size_override hint, all four must hold. _GPT_IMAGE_2_MAX_EDGE_PX = 3840 _GPT_IMAGE_2_MAX_PIXELS = 8_294_400 # 4K ceiling (3840 × 2160) _GPT_IMAGE_2_DIM_MULTIPLE = 16 _GPT_IMAGE_2_MAX_EDGE_RATIO = 3.0 # SSOT for gpt-image-2 pricing in the Recoil engine. Documented per-(size, # quality) tariff in USD per image. Source: fal.ai blog "GPT Image 2 Review" # 2026-05 (verified 2026-05-28). Used to ESTIMATE gpt-image-2 cost because # fal's /edit + /t2i endpoints do not expose cost in the response body or # headers (verified via fal docs 2026-05-26). # # Single source of truth: this table. model_profiles.json::gpt-image-2 points # here via cost_tariff_ssot. Any other site that quotes gpt-image-2 prices # (CLI help text, docs, memory notes) is informational and must be regen'd # from THIS table — never from independent recall. If fal updates pricing, # update this table and `tariff_table_version` below; everything else flows. # # The original estimator (2026-05-26) scaled linearly from a 1024² base by # pixel count — that overshot real pricing by ~4× at 4K because OpenAI/fal # price by output tokens, which scale sub-linearly with pixel count. The # 2026-05-28 rebuild uses the actual published tariff table for documented # sizes and rounds custom sizes UP to the nearest documented size by pixel # count (conservative — never under-estimates). _GPT_IMAGE_2_TARIFF_USD: dict[tuple[int, int], dict[str, float]] = { (1024, 768): {"low": 0.01, "medium": 0.04, "high": 0.15}, (1024, 1024): {"low": 0.01, "medium": 0.06, "high": 0.22}, (1024, 1536): {"low": 0.01, "medium": 0.05, "high": 0.17}, (1536, 1024): {"low": 0.01, "medium": 0.05, "high": 0.17}, # mirror of 1024×1536 (1920, 1080): {"low": 0.01, "medium": 0.04, "high": 0.16}, (2560, 1440): {"low": 0.01, "medium": 0.06, "high": 0.23}, (3840, 2160): {"low": 0.02, "medium": 0.11, "high": 0.41}, } _GPT_IMAGE_2_FALLBACK_TARIFF = _GPT_IMAGE_2_TARIFF_USD[(1024, 1024)] _GPT_IMAGE_2_POLL_INTERVAL_S = 2.0 # 600s = 10 min. Multi-ref i2i edits at high quality can legitimately take # 3-5+ minutes per OpenAI's published "up to 11 min" Thinking-mode runs # (the consult flagged this; field guide notes 2+ min is normal for heavy # generations). 180s was too aggressive — fal accepts the job, our adapter # gives up before fal finishes, work is wasted. Bumped 2026-05-25 after # the first multi-ref live fire surfaced the timeout. _GPT_IMAGE_2_POLL_TIMEOUT_S = 600.0 def _infer_gpt_image_2_action(payload: "UnifiedVideoPayload") -> str: """Discriminate t2i vs edit by typed payload field.""" if payload.reference_images: return "gpt_image_2_edit" return "gpt_image_2_t2i" def _resolve_gpt_image_2_size(aspect_ratio) -> str: """1:1 → 1024x1024, 9:16 → 1024x1536, 16:9 → 1536x1024. Default 1:1.""" return _GPT_IMAGE_2_SIZE_MAP.get(aspect_ratio or "1:1", "1024x1024") def _validate_gpt_image_2_size_override(size: str) -> str | None: """Validate a custom size string against the gpt-image-2 API constraints. Returns the canonical "WIDTHxHEIGHT" string if valid, else None. Constraints (OpenAI public docs 2026-05): both dimensions integer and multiples of 16; each ≤ 3840 px; total pixels ≤ 8,294,400; long-edge to short-edge ratio ≤ 3:1. Anything outside falls back to the preset map rather than risking a 4xx from the upstream. """ if not isinstance(size, str): return None parts = size.lower().split("x") if len(parts) != 2: return None try: w, h = int(parts[0]), int(parts[1]) except ValueError: return None if w <= 0 or h <= 0: return None if max(w, h) > _GPT_IMAGE_2_MAX_EDGE_PX: return None if w * h > _GPT_IMAGE_2_MAX_PIXELS: return None if w % _GPT_IMAGE_2_DIM_MULTIPLE or h % _GPT_IMAGE_2_DIM_MULTIPLE: return None long_edge, short_edge = max(w, h), min(w, h) if long_edge / short_edge > _GPT_IMAGE_2_MAX_EDGE_RATIO: return None return f"{w}x{h}" def _resolve_gpt_image_2_size_with_hints(payload: "UnifiedVideoPayload") -> str: """Read size from payload.hints.size_override (validated), else fall back to the aspect-ratio preset map. Hints win when valid; invalid hints fall through silently so callers can't pin a known-bad size by mistake.""" override = coerce_to_dict(getattr(payload, "hints", None)).get("size_override") if override: validated = _validate_gpt_image_2_size_override(str(override)) if validated: return validated logger.warning( "fal adapter: gpt-image-2 size_override %r failed validation; " "falling back to aspect_ratio preset.", override, ) return _resolve_gpt_image_2_size(payload.aspect_ratio) def _resolve_gpt_image_2_quality(payload: "UnifiedVideoPayload") -> str: q = coerce_to_dict(getattr(payload, "hints", None)).get("quality") return q or getattr(payload, "quality", None) or _GPT_IMAGE_2_DEFAULT_QUALITY def _estimate_gpt_image_2_cost(quality: str, size: str) -> float: """Estimate per-call cost from the (size, quality) tariff table. Fal does not expose cost for gpt-image-2 (verified 2026-05-26 against the /edit endpoint API docs — response body has only the images array; no cost/usage/billing headers). The downstream receipt would otherwise show cost_usd=0.0 or fall back to the static medium-quality $0.08 placeholder in model_profiles. This estimator looks up the published fal.ai tariff for documented (size, quality) pairs. For CUSTOM sizes not in the table (e.g. the new size_override path's 2048×1152), round UP to the nearest documented size by pixel count. Conservative: never under-estimates real cost. Why not linear pixel scaling: the original 2026-05-26 implementation scaled cost = base × (pixels / 1024²) which overshot real pricing by ~4× at 4K. OpenAI/fal charge by output tokens, which scale sub- linearly with pixel count, so each documented size has its own price rather than interpolating from a base. """ q_key = (quality or _GPT_IMAGE_2_DEFAULT_QUALITY).lower() if q_key not in {"low", "medium", "high"}: # Mirrors _resolve_gpt_image_2_size_with_hints's invalid-override # WARNING — keeps the size + quality validation symmetric. Silent # fallback would let an OpenAI tier change (e.g. 'auto') drift the # receipt cost from reality with no operator signal. logger.warning( "fal adapter: gpt-image-2 quality %r not in tariff table " "(low/medium/high); estimating at '%s'.", quality, _GPT_IMAGE_2_DEFAULT_QUALITY, ) q_key = _GPT_IMAGE_2_DEFAULT_QUALITY try: w, h = (int(x) for x in size.lower().split("x")) pixels = w * h if pixels <= 0: raise ValueError except (AttributeError, ValueError, TypeError): return _GPT_IMAGE_2_FALLBACK_TARIFF[q_key] # Exact match on documented size? Use it. tariff = _GPT_IMAGE_2_TARIFF_USD.get((w, h)) if tariff is not None: return tariff[q_key] # Custom size — pick the documented size with the smallest pixel count # that is >= the requested pixel count. (Round UP. Conservative.) # If the requested size exceeds the largest documented size (>4K, which # the validator forbids), fall back to the largest documented tier. candidates = sorted(_GPT_IMAGE_2_TARIFF_USD.items(), key=lambda kv: kv[0][0] * kv[0][1]) for (cw, ch), t in candidates: if cw * ch >= pixels: return t[q_key] # Beyond 4K — return the largest tier. return candidates[-1][1][q_key] def _gpt_image_2_size_str(body: dict) -> str: """Reconstruct 'WxH' from the wire body's image_size object — used by downstream consumers (cost estimator, log lines, receipt metadata) so they don't have to know the wire shape.""" img = body.get("image_size") or {} if isinstance(img, dict) and "width" in img and "height" in img: return f"{img['width']}x{img['height']}" # Preset string fallback (e.g. 'square_hd'); cost lookup won't match # but we surface the literal preset name for traceability. return str(img) if img else "unknown" def _build_gpt_image_2_body(payload: "UnifiedVideoPayload") -> tuple: """Assemble (endpoint, body) for a gpt-image-2 submit. Critical wire-shape: fal's gpt-image-2 endpoints take ``image_size`` as EITHER a preset string (``square_hd`` / ``portrait_16_9`` / ``auto`` etc.) OR an object ``{"width": W, "height": H}`` — NOT a ``"WxH"`` string. The pre-2026-05-28 adapter sent ``body["size"] = "1024x1024"`` (wrong field name + wrong shape), which fal silently ignored and defaulted to ``landscape_4_3`` (1024×768). Every gpt-image-2 fire from 2026-05-19 through 2026-05-28 12:57 UTC actually rendered at 1024×768 regardless of the size requested. JT confirmed via PIL metadata on v1 + v2 fires. Fixed by switching to ``image_size: {width, height}``. """ action_key = _infer_gpt_image_2_action(payload) endpoint = _GPT_IMAGE_2_ACTION_MAP[action_key] size = _resolve_gpt_image_2_size_with_hints(payload) quality = _resolve_gpt_image_2_quality(payload) try: _w, _h = (int(x) for x in size.lower().split("x")) except (AttributeError, ValueError, TypeError): # Defensive — resolver always returns "WxH" but fail safely if not. _w, _h = 1024, 1024 body = { "prompt": payload.prompt, "image_size": {"width": _w, "height": _h}, "quality": quality, "num_images": 1, "output_format": "png", } if action_key == "gpt_image_2_edit": # gpt-image-2/image-to-image accepts up to 16 reference images via the # image_urls array (per fal.ai docs + 2026-05-25 Gemini consult). The # prior [:1] cap was a defensive artifact from the 2026-05-19 wire # build that paired with model_profiles.multi_ref_supported=false # (already corrected to true). Removed 2026-05-25 — was silently # dropping refs 2-N, producing dramatically worse identity anchoring # than the model is actually capable of. _GPT_IMAGE_2_MAX_REFS = 16 uploaded = [] for ref in (payload.reference_images or [])[:_GPT_IMAGE_2_MAX_REFS]: if isinstance(ref, str) and ( ref.startswith("http") or ref.startswith("fal://") ): uploaded.append(ref) elif isinstance(ref, str) and os.path.isfile(ref): if not _HAS_FAL: raise RuntimeError("fal_client not installed — cannot upload ref") uploaded.append(_fal_client.upload_file(ref)) elif isinstance(ref, (bytes, bytearray)): uploaded.append(_upload_bytes_to_fal(bytes(ref))) if uploaded: body["image_urls"] = uploaded return endpoint, body def _extract_fal_billing_cost_usd(response) -> float: """Pull cost_usd from fal response. Returns 0.0 on any missing/malformed field — never raises. DEPRECATED for gpt-image-2 (2026-05-26): fal does not expose this endpoint's cost in the response body or headers, so calls would always return 0.0 here and emit a misleading WARNING. The gpt-image-2 path now uses ``_estimate_gpt_image_2_cost`` instead. This helper remains live for other fal-routed models (seeddance video etc.) where billing IS in the response. If you wire a new fal endpoint, verify the response shape exposes cost before calling this — otherwise build an estimator. """ if not isinstance(response, dict): return 0.0 for key in ("cost_usd", "total_cost_usd"): v = response.get(key) if isinstance(v, (int, float)): return float(v) billing = response.get("billing") if isinstance(billing, dict): for key in ("cost_usd", "total_cost_usd"): v = billing.get(key) if isinstance(v, (int, float)): return float(v) logger.warning( "fal adapter: response missing cost_usd / total_cost_usd " "(flat or under .billing). cost_usd defaulted to 0.0." ) return 0.0 def _resolve_model_path(tier: str, action: str) -> str: if tier.startswith("fast"): return f"bytedance/seedance-2.0/fast/{action}" return f"bytedance/seedance-2.0/{action}" def _tier_resolution(tier: str, fallback: str) -> str: """'fast_480p' -> '480p'; else the payload-supplied value.""" if tier.endswith("_480p"): return "480p" if tier.endswith("_720p"): return "720p" return fallback def _to_bytes(value) -> bytes: """Coerce str (base64) or bytes to bytes.""" if isinstance(value, (bytes, bytearray)): return bytes(value) if isinstance(value, str): return base64.b64decode(value) raise TypeError(f"fal adapter: cannot coerce {type(value).__name__} to bytes") def _infer_action(payload: UnifiedVideoPayload) -> str: if payload.image is not None: return "i2v" if payload.reference_images or payload.reference_videos: return "r2v" return "t2v" # --------------------------------------------------------------- # The adapter # --------------------------------------------------------------- class FalAdapter: provider_id = "fal" supported_models = ["seeddance-2.0", "gpt-image-2"] auth_env_var = "FAL_KEY" base_url = _QUEUE_BASE max_prompt_chars = None status = "primary" capabilities = { "t2v": True, "i2v": True, "r2v": True, "end_frame": True, # fal.ai supports end_image_url "audio": True, "negative_prompt": True, "resolution_480p": True, "resolution_720p": True, "resolution_1080p": False, } def __init__(self): try: self.transport = _fal_transport.FalTransport(auth_env_var=self.auth_env_var) except TypeError: # Existing unit tests monkeypatch FalTransport with no-arg fakes. self.transport = _fal_transport.FalTransport() # ---- headers ---- def _headers(self) -> dict: key = os.environ.get(self.auth_env_var) if not key: raise RuntimeError(f"{self.auth_env_var} environment variable not set.") return { "Authorization": f"Key {key}", "Content-Type": "application/json", } # ---- submit ---- def build_submit(self, payload: UnifiedVideoPayload, tier: str) -> SubmitRequest: action_key = _infer_action(payload) action = _ACTION_MAP[action_key] model_path = _resolve_model_path(tier, action) resolution = _tier_resolution(tier, payload.resolution) body: dict = { "prompt": payload.prompt, "duration": str(payload.duration_s), "resolution": resolution, } if payload.aspect_ratio: body["aspect_ratio"] = payload.aspect_ratio # Start frame if payload.image is not None: image_url = _upload_bytes_to_fal(_to_bytes(payload.image)) body["image_url"] = image_url logger.info("fal adapter: uploaded start frame -> %s", image_url[:80]) # End frame if payload.image_tail is not None: end_url = _upload_bytes_to_fal(_to_bytes(payload.image_tail)) body["end_image_url"] = end_url logger.info("fal adapter: uploaded end frame -> %s", end_url[:80]) # Reference images if payload.reference_images: uploaded: list[str] = [] for ref in payload.reference_images: if isinstance(ref, str) and ( ref.startswith("http") or ref.startswith("fal://") ): uploaded.append(ref) elif isinstance(ref, str) and os.path.isfile(ref): if not _HAS_FAL: raise RuntimeError( "fal_client not installed — cannot upload ref" ) uploaded.append(_fal_client.upload_file(ref)) elif isinstance(ref, (bytes, bytearray)): uploaded.append(_upload_bytes_to_fal(bytes(ref))) else: logger.warning( "fal adapter: skipping invalid ref %r", str(ref)[:60] ) if uploaded: body["image_urls"] = uploaded # Reference videos (Seedance R2V, max 3 per fal.ai schema) if payload.reference_videos: uploaded_videos: list[str] = [] for ref in payload.reference_videos: if isinstance(ref, str) and ( ref.startswith("http") or ref.startswith("fal://") ): uploaded_videos.append(ref) elif isinstance(ref, str) and os.path.isfile(ref): if not _HAS_FAL: raise RuntimeError( "fal_client not installed — cannot upload video ref" ) uploaded_videos.append(_fal_client.upload_file(ref)) logger.info( "fal adapter: uploaded video ref %s", os.path.basename(ref) ) else: logger.warning( "fal adapter: skipping invalid video ref %r", str(ref)[:60] ) if uploaded_videos: body["video_urls"] = uploaded_videos # Reference audios (Seedance R2V, native audio channel — see # pipeline-learnings §17c). Mirror the image_urls / video_urls pattern. if payload.reference_audios: uploaded_audios: list[str] = [] for ref in payload.reference_audios: if isinstance(ref, str) and ( ref.startswith("http") or ref.startswith("fal://") ): uploaded_audios.append(ref) elif isinstance(ref, str) and os.path.isfile(ref): if not _HAS_FAL: raise RuntimeError( "fal_client not installed — cannot upload audio ref" ) uploaded_audios.append(_fal_client.upload_file(ref)) logger.info( "fal adapter: uploaded audio ref %s", os.path.basename(ref) ) else: logger.warning( "fal adapter: skipping invalid audio ref %r", str(ref)[:60] ) if uploaded_audios: body["audio_urls"] = uploaded_audios if payload.generate_audio: body["generate_audio"] = True if payload.negative_prompt: body["negative_prompt"] = payload.negative_prompt url = f"{_QUEUE_BASE}/{model_path}" return SubmitRequest(method="POST", url=url, headers=self._headers(), body=body) def parse_submit( self, resp: dict, payload: UnifiedVideoPayload, tier: str ) -> ProviderJob: request_id = resp.get("request_id") if not request_id: raise RuntimeError( f"fal adapter: no request_id in submit response: " f"{resp.get('detail', resp)}" ) action_key = _infer_action(payload) action = _ACTION_MAP[action_key] model_path = _resolve_model_path(tier, action) resolution = _tier_resolution(tier, payload.resolution) return ProviderJob( provider_id=self.provider_id, model_id=payload.model_id, native_id=request_id, tier=tier, duration_s=payload.duration_s, resolution=resolution, native_state={ "model_path": model_path, "status_url": resp.get("status_url") or f"{_QUEUE_BASE}/{model_path}/requests/{request_id}/status", "response_url": resp.get("response_url") or f"{_QUEUE_BASE}/{model_path}/requests/{request_id}", }, ) # ---- poll ---- def build_poll(self, job: ProviderJob) -> PollRequest: return PollRequest( method="GET", url=job.native_state["status_url"], headers=self._headers(), ) def parse_poll(self, resp: dict, job: ProviderJob) -> PollResult: status = (resp.get("status") or "unknown").upper() if status == "COMPLETED": return PollResult(status="COMPLETED", raw=resp) if status == "FAILED": return PollResult( status="FAILED", error=resp.get("error") or "fal.ai generation failed", raw=resp, ) return PollResult(status="IN_PROGRESS", raw=resp) # ---- result fetch ---- def build_result_fetch(self, job: ProviderJob) -> Optional[PollRequest]: return PollRequest( method="GET", url=job.native_state["response_url"], headers=self._headers(), ) def parse_result(self, resp: dict, job: ProviderJob) -> PollResult: video = resp.get("video") or {} audio = resp.get("audio") or {} # Phase 7 (Bug L): normalize seed / request_id / inference_billed_usd # to top-level keys on the raw dict so downstream metadata harvesting # (video_model_client._finalize_completed) finds them consistently # regardless of whether fal nests them under `video` or returns them # at the top level. Previously `seed` was silently swallowed. normalized = dict(resp) if "seed" not in normalized: nested_seed = video.get("seed") if isinstance(video, dict) else None if nested_seed is not None: normalized["seed"] = nested_seed if "request_id" not in normalized and job is not None: normalized["request_id"] = job.native_id billing = resp.get("billing") if isinstance(resp.get("billing"), dict) else {} if "inference_billed_usd" not in normalized: for k in ("inference_billed_usd", "cost_usd", "total_cost_usd"): v = billing.get(k) if billing else resp.get(k) if isinstance(v, (int, float)): normalized["inference_billed_usd"] = float(v) break return PollResult( status="COMPLETED", video_url=video.get("url"), audio_url=audio.get("url"), observed_cost=None, # fal doesn't return billing inline raw=normalized, ) # ---- cost ---- def compute_cost(self, duration_s: float, tier: str, profile: dict) -> float: providers = (profile or {}).get("providers", {}) fal_block = providers.get("fal.ai") or providers.get("fal") or {} tiers = fal_block.get("tiers", {}) rate = (tiers.get(tier) or {}).get("cost_per_second") if rate is None: # Legacy fallback — preserve prior behavior. if tier.startswith("fast") and tier.endswith("_480p"): rate = profile.get("cost_per_second_fast_480p", 0.111) elif tier.startswith("fast"): rate = profile.get("cost_per_second_fast", 0.2419) else: rate = profile.get("cost_per_second", 0.3034) return float(rate) * float(duration_s) def direct_submit_image(self, payload: "UnifiedVideoPayload") -> dict: """Synchronous image generation for gpt-image-2 via fal.ai. Wraps the queue→status→result HTTP lifecycle. Returns: - image_bytes: PNG bytes - native_id: fal request id - cost_usd: float from fal billing (0.0 if missing) - metadata: {endpoint, size, quality, cost_source} """ if payload.model_id != "gpt-image-2": raise NotImplementedError( f"FalAdapter.direct_submit_image not yet wired for " f"model_id={payload.model_id!r}. v1 supports gpt-image-2 only." ) endpoint, body = _build_gpt_image_2_body(payload) _size_str = _gpt_image_2_size_str(body) # PRE-SUBMIT cost-cap enforcement. Estimate cost from the resolved # (quality, size), compare to the profile cap, and raise BEFORE # submitting if exceeded. The estimator's accuracy is ~10% so we # use the cap as a hard ceiling rather than adding a buffer — bias # toward refusing too much over spending too much. _estimated_cost = _estimate_gpt_image_2_cost(body["quality"], _size_str) try: from recoil.core import model_profiles _profile = model_profiles.get_profile(payload.model_id) or {} _cap = _profile.get("max_cost_per_shot_usd") except Exception: # noqa: BLE001 — profile lookup is opportunistic _cap = None if _cap is not None and _estimated_cost > float(_cap): raise RuntimeError( f"fal adapter: gpt-image-2 call would cost ~${_estimated_cost:.4f} " f"(quality={body['quality']!r}, size={_size_str!r}) — exceeds " f"max_cost_per_shot_usd=${float(_cap):.2f} in model_profiles. " f"Either lower --quality, drop --size override, or raise the cap " f"in recoil/config/model_profiles.json::gpt-image-2 if intentional." ) transport = _fal_transport.FalTransport() submit_resp = transport.submit_queue(endpoint, body) request_id = submit_resp.get("request_id") if not request_id: raise RuntimeError( f"fal adapter: no request_id in gpt-image-2 submit response: " f"{submit_resp.get('detail', submit_resp)}" ) status_url = ( submit_resp.get("status_url") or f"{_QUEUE_BASE}/{endpoint}/requests/{request_id}/status" ) response_url = ( submit_resp.get("response_url") or f"{_QUEUE_BASE}/{endpoint}/requests/{request_id}" ) deadline = time.time() + _GPT_IMAGE_2_POLL_TIMEOUT_S poll_interval = _GPT_IMAGE_2_POLL_INTERVAL_S while True: if time.time() > deadline: raise RuntimeError( f"fal adapter: gpt-image-2 poll timeout after " f"{_GPT_IMAGE_2_POLL_TIMEOUT_S}s " f"(request_id={request_id}, response_url={response_url})" ) try: status_resp = transport.poll_status(status_url) except Exception as e: logger.warning( "fal adapter: gpt-image-2 poll error " "(request_id=%s, response_url=%s): %s", request_id, response_url, e, ) time.sleep(poll_interval) poll_interval = min(poll_interval * 1.5, 30.0) continue status = (status_resp.get("status") or "unknown").upper() if status == "COMPLETED": break if status == "FAILED": raise RuntimeError( f"fal adapter: gpt-image-2 FAILED (request_id={request_id}): " f"{status_resp.get('error') or status_resp}" ) time.sleep(_GPT_IMAGE_2_POLL_INTERVAL_S) result_resp = transport.fetch_result(response_url) images = result_resp.get("images") or [] if not images: single = result_resp.get("image") if isinstance(single, dict) and single.get("url"): images = [single] if not images or not isinstance(images[0], dict): raise RuntimeError( f"fal adapter: gpt-image-2 result has no images " f"(request_id={request_id}): {result_resp}" ) image_url = images[0].get("url") if not image_url: raise RuntimeError( f"fal adapter: gpt-image-2 result image has no url " f"(request_id={request_id}): {images[0]}" ) try: with urllib.request.urlopen(image_url, timeout=30) as resp: image_bytes = resp.read() except urllib.error.URLError as e: raise RuntimeError( f"fal adapter: gpt-image-2 image download failed ({image_url[:120]}): {e}" ) from e if not image_bytes: raise RuntimeError( f"fal adapter: gpt-image-2 downloaded image is empty ({image_url[:120]})" ) # gpt-image-2: fal does not expose cost in /edit or /t2i response # body or headers (verified 2026-05-26 against fal docs). Calling # _extract_fal_billing_cost_usd would always log a misleading # "missing cost" WARNING for this endpoint. Use the per-tier # estimator instead — grounded in OpenAI's published pricing and # accurate to within ~10% of actual. cost_usd = _estimate_gpt_image_2_cost(body["quality"], _size_str) cost_source = "estimated_from_tariff" logger.info( "fal adapter: gpt-image-2 OK (request_id=%s, size=%s, " "quality=%s, cost_usd=%.4f, source=%s)", request_id, _size_str, body["quality"], cost_usd, cost_source, ) return { "image_bytes": image_bytes, "native_id": request_id, "cost_usd": cost_usd, "metadata": { "endpoint": endpoint, "size": _size_str, "quality": body["quality"], "cost_source": cost_source, # Reconciliation data — downstream tools can compare # estimator output against real OpenAI invoices by grouping # on (quality, size). Preserves the "we don't know vs we # measured" signal that fal stripped by not exposing cost. "cost_estimator_inputs": { "quality": body["quality"], "size": _size_str, "tariff_table_version": "2026-05-28-fal-blog", "tariff_table_source": "fal.ai/learn/tools/gpt-image-2-review", }, }, } ADAPTER = FalAdapter() __all__ = ["FalAdapter", "ADAPTER"]