Fal Ai Image

{
  "grok-imagine-image-t2i": {
    "provider": "fal",
    "family": "grok-imagine-image",
    "endpoint_id": "xai/grok-imagine-image",
    "task_type": "text-to-image",
    "download_keys": ["images[0].url"],
    "defaults": {
      "num_images": 1,
      "aspect_ratio": "1:1",
      "output_format": "png"
    }
  },
  "grok-imagine-image-edit": {
    "provider": "fal",
    "family": "grok-imagine-image",
    "endpoint_id": "xai/grok-imagine-image/edit",
    "task_type": "image-edit",
    "input_image_field": "image_urls",
    "download_keys": ["images[0].url"],
    "defaults": {
      "num_images": 1,
      "output_format": "png"
    }
  },
  "nano-banana-2-t2i": {
    "provider": "fal",
    "family": "nano-banana-2",
    "endpoint_id": "fal-ai/nano-banana-2",
    "task_type": "text-to-image",
    "download_keys": ["images[0].url"],
    "defaults": {
      "num_images": 1,
      "aspect_ratio": "1:1",
      "output_format": "png",
      "resolution": "1K",
      "safety_tolerance": "4"
    }
  },
  "nano-banana-2-edit": {
    "provider": "fal",
    "family": "nano-banana-2",
    "endpoint_id": "fal-ai/nano-banana-2/edit",
    "task_type": "image-edit",
    "input_image_field": "image_urls",
    "download_keys": ["images[0].url"],
    "defaults": {
      "num_images": 1,
      "aspect_ratio": "auto",
      "output_format": "png",
      "resolution": "1K",
      "safety_tolerance": "4"
    }
  },
  "nano-banana-pro-t2i": {
    "provider": "fal",
    "family": "nano-banana-pro",
    "endpoint_id": "fal-ai/nano-banana-pro",
    "task_type": "text-to-image",
    "download_keys": ["images[0].url"],
    "defaults": {
      "num_images": 1,
      "aspect_ratio": "1:1",
      "output_format": "png",
      "resolution": "1K",
      "safety_tolerance": "4"
    }
  },
  "nano-banana-pro-edit": {
    "provider": "fal",
    "family": "nano-banana-pro",
    "endpoint_id": "fal-ai/nano-banana-pro/edit",
    "task_type": "image-edit",
    "input_image_field": "image_urls",
    "download_keys": ["images[0].url"],
    "defaults": {
      "num_images": 1,
      "aspect_ratio": "auto",
      "output_format": "png",
      "resolution": "1K",
      "safety_tolerance": "4"
    }
  },
  "gpt-image-1.5-t2i": {
    "provider": "fal",
    "family": "gpt-image-1.5",
    "endpoint_id": "fal-ai/gpt-image-1.5",
    "task_type": "text-to-image",
    "download_keys": ["images[0].url"],
    "defaults": {
      "num_images": 1,
      "image_size": "1024x1024",
      "background": "auto",
      "output_format": "png",
      "quality": "medium"
    }
  },
  "gpt-image-1.5-edit": {
    "provider": "fal",
    "family": "gpt-image-1.5",
    "endpoint_id": "fal-ai/gpt-image-1.5/edit",
    "task_type": "image-edit",
    "input_image_field": "image_urls",
    "download_keys": ["images[0].url"],
    "defaults": {
      "num_images": 1,
      "image_size": "auto",
      "background": "auto",
      "output_format": "png",
      "quality": "medium"
    }
  }
}

Keep the same pirate character identity and pose structure, but refine the image into a cleaner game-reference illustration.
Preserve the subject, costume, proportions, and silhouette.
Transparent or simple clean background only.
No extra characters, no props added unless explicitly requested, no scene expansion.

Create a clean standalone pirate character concept based on the game's existing pirate identity.
Full body, readable silhouette, transparent background, centered composition, no extra characters, no text, no environment.
Keep the design usable as a game-reference concept rather than a cinematic poster.

#!/usr/bin/env python3
from __future__ import annotations

import base64
import csv
import hashlib
import json
import mimetypes
import os
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


QUEUE_ROOT = "https://queue.fal.run"
PLATFORM_ROOT = "https://api.fal.ai/v1"
REPO_ROOT = Path(__file__).resolve().parents[4]
SKILL_ROOT = Path(__file__).resolve().parents[1]
PRESETS_PATH = SKILL_ROOT / "assets" / "model-presets.json"


@dataclass
class HttpJsonResponse:
    status_code: int
    headers: dict[str, str]
    payload: dict[str, Any]


def now_utc_iso() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def timestamp_slug() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%d-%H%M%S")


def require_fal_key() -> str:
    api_key = os.environ.get("FAL_KEY") or os.environ.get("FAL_API_KEY")
    if not api_key:
        raise SystemExit("FAL_KEY or FAL_API_KEY is required")
    return api_key


def load_presets() -> dict[str, dict[str, Any]]:
    return json.loads(PRESETS_PATH.read_text(encoding="utf-8"))


def prompt_sha256(prompt: str) -> str:
    return hashlib.sha256(prompt.encode("utf-8")).hexdigest()


def ensure_parent(path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)


def write_json(path: Path, payload: Any) -> None:
    ensure_parent(path)
    path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")


def append_jsonl(path: Path, row: dict[str, Any]) -> None:
    ensure_parent(path)
    with path.open("a", encoding="utf-8") as handle:
        handle.write(json.dumps(row, ensure_ascii=False) + "\n")


def write_csv(path: Path, rows: list[dict[str, Any]], fieldnames: list[str]) -> None:
    ensure_parent(path)
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        for row in rows:
            writer.writerow({name: row.get(name, "") for name in fieldnames})


def load_jsonl(path: Path) -> list[dict[str, Any]]:
    if not path.exists():
        return []
    rows: list[dict[str, Any]] = []
    for line in path.read_text(encoding="utf-8").splitlines():
        stripped = line.strip()
        if stripped:
            rows.append(json.loads(stripped))
    return rows


def normalize_headers(headers: Any) -> dict[str, str]:
    normalized: dict[str, str] = {}
    for key, value in headers.items():
        normalized[str(key).lower()] = str(value)
    return normalized


def fal_headers(api_key: str, extra_headers: dict[str, str] | None = None, *, json_content: bool = False) -> dict[str, str]:
    headers = {
        "Authorization": f"Key {api_key}",
        "Accept": "application/json",
    }
    if json_content:
        headers["Content-Type"] = "application/json"
    if extra_headers:
        headers.update(extra_headers)
    return headers


def data_uri_for_file(path: Path) -> str:
    if not path.exists():
        raise SystemExit(f"Image file not found: {path}")
    mime = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
    encoded = base64.b64encode(path.read_bytes()).decode("ascii")
    return f"data:{mime};base64,{encoded}"


def _json_request(method: str, url: str, api_key: str, payload: Any | None = None, headers: dict[str, str] | None = None) -> HttpJsonResponse:
    request_headers = fal_headers(api_key, headers, json_content=payload is not None)
    data = json.dumps(payload).encode("utf-8") if payload is not None else None
    request = urllib.request.Request(url, data=data, headers=request_headers, method=method)
    try:
        with urllib.request.urlopen(request) as response:
            body = response.read().decode("utf-8")
            response_headers = normalize_headers(response.headers)
            parsed = json.loads(body) if body else {}
            return HttpJsonResponse(response.getcode(), response_headers, parsed)
    except urllib.error.HTTPError as exc:
        error_text = exc.read().decode("utf-8", errors="replace")
        raise SystemExit(f"fal API error {exc.code}: {error_text}") from exc
    except urllib.error.URLError as exc:
        raise SystemExit(f"Network error: {exc}") from exc


def queue_submit(endpoint_id: str, api_key: str, payload: dict[str, Any], headers: dict[str, str] | None = None) -> HttpJsonResponse:
    return _json_request("POST", f"{QUEUE_ROOT}/{endpoint_id.strip('/')}", api_key, payload=payload, headers=headers)


def queue_get_by_url(url: str, api_key: str, headers: dict[str, str] | None = None, *, query: dict[str, Any] | None = None) -> HttpJsonResponse:
    encoded_query = ""
    if query:
        items: list[tuple[str, str]] = []
        for key, value in query.items():
            if value is None:
                continue
            if isinstance(value, list):
                for item in value:
                    items.append((key, str(item)))
            else:
                items.append((key, str(value)))
        encoded_query = f"?{urllib.parse.urlencode(items)}" if items else ""
    return _json_request("GET", f"{url}{encoded_query}", api_key, headers=headers)


def queue_status(endpoint_id: str, request_id: str, api_key: str, headers: dict[str, str] | None = None, *, logs: bool = True) -> HttpJsonResponse:
    query = "?logs=1" if logs else ""
    return _json_request(
        "GET",
        f"{QUEUE_ROOT}/{endpoint_id.strip('/')}/requests/{request_id}/status{query}",
        api_key,
        headers=headers,
    )


def queue_result(endpoint_id: str, request_id: str, api_key: str, headers: dict[str, str] | None = None) -> HttpJsonResponse:
    return _json_request(
        "GET",
        f"{QUEUE_ROOT}/{endpoint_id.strip('/')}/requests/{request_id}",
        api_key,
        headers=headers,
    )


def platform_get(path: str, api_key: str, query: dict[str, Any] | None = None) -> HttpJsonResponse:
    encoded_query = ""
    if query:
        items: list[tuple[str, str]] = []
        for key, value in query.items():
            if value is None:
                continue
            if isinstance(value, list):
                for item in value:
                    items.append((key, str(item)))
            else:
                items.append((key, str(value)))
        encoded_query = f"?{urllib.parse.urlencode(items)}" if items else ""
    return _json_request("GET", f"{PLATFORM_ROOT}{path}{encoded_query}", api_key)


def platform_post(path: str, api_key: str, payload: dict[str, Any]) -> HttpJsonResponse:
    return _json_request("POST", f"{PLATFORM_ROOT}{path}", api_key, payload=payload)


def read_text(path: Path) -> str:
    return path.read_text(encoding="utf-8").strip()


def repo_relative(path: Path) -> str:
    try:
        return path.resolve().relative_to(REPO_ROOT).as_posix()
    except ValueError:
        return path.resolve().as_posix()


def download_file(url: str, out_path: Path) -> None:
    request = urllib.request.Request(url, method="GET")
    try:
        with urllib.request.urlopen(request) as response:
            ensure_parent(out_path)
            out_path.write_bytes(response.read())
    except urllib.error.HTTPError as exc:
        error_text = exc.read().decode("utf-8", errors="replace")
        raise SystemExit(f"Download failed {exc.code}: {error_text}") from exc
    except urllib.error.URLError as exc:
        raise SystemExit(f"Download failed: {exc}") from exc


def _walk_media(value: Any, path: str = "") -> list[dict[str, Any]]:
    found: list[dict[str, Any]] = []
    if isinstance(value, dict):
        if isinstance(value.get("url"), str):
            found.append(
                {
                    "path": path or "url",
                    "url": value["url"],
                    "content_type": value.get("content_type"),
                    "file_name": value.get("file_name"),
                }
            )
        for key, child in value.items():
            child_path = f"{path}.{key}" if path else key
            found.extend(_walk_media(child, child_path))
    elif isinstance(value, list):
        for index, child in enumerate(value):
            child_path = f"{path}[{index}]"
            found.extend(_walk_media(child, child_path))
    return found


def collect_media_outputs(payload: dict[str, Any]) -> list[dict[str, Any]]:
    return _walk_media(payload)


def default_platform_headers(*, store_io: bool = True, disable_fallback: bool = True) -> dict[str, str]:
    headers: dict[str, str] = {}
    if store_io:
        headers["X-Fal-Store-IO"] = "1"
    if disable_fallback:
        headers["x-app-fal-disable-fallback"] = "true"
    return headers


def build_unit_price_estimate_payload(endpoint_id: str, unit_quantity: float) -> dict[str, Any]:
    return {
        "estimate_type": "unit_price",
        "endpoints": {
            endpoint_id: {
                "unit_quantity": unit_quantity,
            }
        },
    }


def coerce_json_object(text: str | None) -> dict[str, Any]:
    if not text:
        return {}
    payload = json.loads(text)
    if not isinstance(payload, dict):
        raise SystemExit("Expected a JSON object")
    return payload

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Any

from _fal_common import (
    append_jsonl,
    load_jsonl,
    now_utc_iso,
    repo_relative,
    timestamp_slug,
    write_csv,
    write_json,
)
from fal_queue_image_run import parse_args as parse_runner_args
from fal_queue_image_run import run_image_job


LEDGER_FIELDS = [
    "timestamp",
    "task_slug",
    "model_alias",
    "endpoint_id",
    "request_id",
    "status",
    "estimated_cost",
    "estimated_cost_method",
    "cost_currency",
    "billable_units_header",
    "output_files",
    "output_urls",
    "run_manifest",
]


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Run the same fal image task across a configured model matrix.")
    parser.add_argument("--config", type=Path, required=True, help="Path to the experiment config JSON file.")
    parser.add_argument("--timestamp", default=None, help="Optional fixed batch timestamp.")
    parser.add_argument("--dry-run", action="store_true", help="Resolve runs without submitting them.")
    return parser.parse_args()


def _load_config(path: Path) -> dict[str, Any]:
    payload = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(payload, dict):
        raise SystemExit("Config must be a JSON object")
    return payload


def _prompt_spec(config: dict[str, Any]) -> tuple[str, str]:
    if bool(config.get("prompt")) == bool(config.get("prompt_file")):
        raise SystemExit("Config must include exactly one of prompt or prompt_file")
    if config.get("prompt"):
        return "prompt", str(config["prompt"])
    return "prompt_file", str(config["prompt_file"])


def _extend_input_flags(runner_args: list[str], config: dict[str, Any]) -> None:
    for item in config.get("input_images", []) or []:
        value = str(item)
        if value.startswith("http://") or value.startswith("https://") or value.startswith("data:"):
            runner_args.extend(["--image-url", value])
        else:
            runner_args.extend(["--image-file", value])


def main() -> None:
    args = parse_args()
    config = _load_config(args.config)
    batch_timestamp = args.timestamp or timestamp_slug()

    batch_dir = Path("experiments/fal-image") / f"{batch_timestamp}-{config['task_slug']}"
    batch_dir.mkdir(parents=True, exist_ok=True)

    tracking = config.get("tracking", {})
    ledger_jsonl = Path(tracking.get("ledger_jsonl", "experiments/fal-image/ledger.jsonl"))
    ledger_csv = Path(tracking.get("ledger_csv", "experiments/fal-image/ledger.csv"))
    prompt_key, prompt_value = _prompt_spec(config)

    batch_payload = {
        "timestamp": batch_timestamp,
        "task_slug": config["task_slug"],
        "created_at": now_utc_iso(),
        "prompt": config.get("prompt"),
        "prompt_file": config.get("prompt_file"),
        "input_images": config.get("input_images", []),
        "models": config["models"],
        "output_root": config["output_root"],
        "config_path": repo_relative(args.config),
    }
    write_json(batch_dir / "batch.json", batch_payload)

    results: list[dict[str, Any]] = []
    for model_alias in config["models"]:
        runner_args = [
            "--model-alias",
            str(model_alias),
            "--task-slug",
            str(config["task_slug"]),
            "--out-dir",
            str(Path(config["output_root"]) / f"{batch_timestamp}-{config['task_slug']}-fal-{model_alias}"),
            "--filename-prefix",
            f"{batch_timestamp}-{config['task_slug']}-{model_alias}",
            f"--{prompt_key.replace('_', '-')}",
            prompt_value,
        ]

        _extend_input_flags(runner_args, config)

        model_overrides = (config.get("model_overrides") or {}).get(model_alias, {})
        for key, value in model_overrides.items():
            flag = f"--{key.replace('_', '-')}"
            runner_args.extend([flag, str(value).lower() if isinstance(value, bool) else str(value)])
        if args.dry_run:
            runner_args.append("--dry-run")

        runner_namespace = parse_runner_args(runner_args)
        manifest = run_image_job(runner_namespace)
        manifest_path = Path(runner_namespace.out_dir) / f"{runner_namespace.filename_prefix}-run.json"
        ledger_row = {
            "timestamp": manifest.get("timestamp"),
            "task_slug": manifest.get("task_slug"),
            "model_alias": manifest.get("model_alias"),
            "endpoint_id": manifest.get("endpoint_id"),
            "request_id": manifest.get("request_id"),
            "status": manifest.get("status"),
            "estimated_cost": (manifest.get("estimated_cost") or {}).get("total_cost") if isinstance(manifest.get("estimated_cost"), dict) else None,
            "estimated_cost_method": manifest.get("estimated_cost_method"),
            "cost_currency": manifest.get("cost_currency"),
            "billable_units_header": manifest.get("billable_units_header"),
            "output_files": "|".join(manifest.get("output_files", [])),
            "output_urls": "|".join(manifest.get("output_urls", [])),
            "run_manifest": repo_relative(manifest_path),
        }
        append_jsonl(ledger_jsonl, ledger_row)
        results.append(ledger_row)

    existing_rows = load_jsonl(ledger_jsonl)
    write_csv(ledger_csv, existing_rows, LEDGER_FIELDS)
    write_json(batch_dir / "results.json", {"rows": results})


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json

from _fal_common import platform_get, platform_post, require_fal_key


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Query fal platform APIs for image models, pricing, usage, and request audit.")
    subparsers = parser.add_subparsers(dest="command", required=True)

    search = subparsers.add_parser("search", help="Search or fetch fal model metadata.")
    search.add_argument("--endpoint-id", action="append", default=None, help="Specific endpoint id. Repeatable.")
    search.add_argument("--query", default=None, help="Free-text model search.")
    search.add_argument("--category", default=None, help="Model category filter.")
    search.add_argument("--status", default=None, help="Model status filter.")
    search.add_argument("--expand", action="append", default=None, help="Expansion values such as openapi-3.0.")
    search.add_argument("--limit", type=int, default=20)

    pricing = subparsers.add_parser("pricing", help="Fetch pricing for one or more endpoints.")
    pricing.add_argument("--endpoint-id", action="append", required=True)

    estimate = subparsers.add_parser("estimate", help="Estimate unit-price cost for one or more endpoints.")
    estimate.add_argument("--endpoint-id", action="append", required=True)
    estimate.add_argument("--unit-quantity", type=float, default=1.0, help="Billing units per endpoint.")

    usage = subparsers.add_parser("usage", help="Fetch usage summaries.")
    usage.add_argument("--endpoint-id", action="append", default=None)
    usage.add_argument("--start", default=None)
    usage.add_argument("--end", default=None)
    usage.add_argument("--timeframe", default=None)
    usage.add_argument("--expand", action="append", default=["summary"])
    usage.add_argument("--limit", type=int, default=50)

    analytics = subparsers.add_parser("analytics", help="Fetch analytics summaries.")
    analytics.add_argument("--endpoint-id", action="append", default=None)
    analytics.add_argument("--start", default=None)
    analytics.add_argument("--end", default=None)
    analytics.add_argument("--timeframe", default=None)
    analytics.add_argument("--expand", action="append", default=["summary", "time_series", "request_count"])
    analytics.add_argument("--limit", type=int, default=50)

    requests = subparsers.add_parser("requests", help="Fetch request audit records by endpoint.")
    requests.add_argument("--endpoint-id", required=True)
    requests.add_argument("--request-id", default=None)
    requests.add_argument("--status", default=None)
    requests.add_argument("--start", default=None)
    requests.add_argument("--end", default=None)
    requests.add_argument("--expand", action="append", default=None)
    requests.add_argument("--limit", type=int, default=20)

    return parser.parse_args()


def main() -> None:
    args = parse_args()
    api_key = require_fal_key()

    if args.command == "search":
        response = platform_get(
            "/models",
            api_key,
            {
                "endpoint_id": args.endpoint_id,
                "q": args.query,
                "category": args.category,
                "status": args.status,
                "expand": args.expand,
                "limit": args.limit,
            },
        )
    elif args.command == "pricing":
        response = platform_get("/models/pricing", api_key, {"endpoint_id": args.endpoint_id})
    elif args.command == "estimate":
        endpoints = {endpoint_id: {"unit_quantity": args.unit_quantity} for endpoint_id in args.endpoint_id}
        response = platform_post("/models/pricing/estimate", api_key, {"estimate_type": "unit_price", "endpoints": endpoints})
    elif args.command == "usage":
        response = platform_get(
            "/models/usage",
            api_key,
            {
                "endpoint_id": args.endpoint_id,
                "start": args.start,
                "end": args.end,
                "timeframe": args.timeframe,
                "expand": args.expand,
                "limit": args.limit,
            },
        )
    elif args.command == "analytics":
        response = platform_get(
            "/models/analytics",
            api_key,
            {
                "endpoint_id": args.endpoint_id,
                "start": args.start,
                "end": args.end,
                "timeframe": args.timeframe,
                "expand": args.expand,
                "limit": args.limit,
            },
        )
    else:
        response = platform_get(
            "/models/requests/by-endpoint",
            api_key,
            {
                "endpoint_id": args.endpoint_id,
                "request_id": args.request_id,
                "status": args.status,
                "start": args.start,
                "end": args.end,
                "expand": args.expand,
                "limit": args.limit,
            },
        )

    print(json.dumps(response.payload, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import os
import time
from pathlib import Path
from typing import Any, Sequence

from _fal_common import (
    build_unit_price_estimate_payload,
    collect_media_outputs,
    coerce_json_object,
    data_uri_for_file,
    default_platform_headers,
    download_file,
    load_presets,
    now_utc_iso,
    platform_get,
    platform_post,
    prompt_sha256,
    queue_get_by_url,
    queue_result,
    queue_status,
    queue_submit,
    read_text,
    repo_relative,
    require_fal_key,
    write_json,
)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Run a fal.ai image queue job and write normalized tracking artifacts.")
    parser.add_argument("--model-alias", default=None, help="Friendly alias from assets/model-presets.json.")
    parser.add_argument("--endpoint-id", default=None, help="Raw fal endpoint id. Overrides preset endpoint.")
    parser.add_argument("--prompt", default=None, help="Prompt text.")
    parser.add_argument("--prompt-file", type=Path, default=None, help="Path to a text file containing the prompt.")
    parser.add_argument("--image-file", type=Path, action="append", default=None, help="Local input image file. Repeat for edits with multiple references.")
    parser.add_argument("--image-url", action="append", default=None, help="Hosted input image URL. Repeatable.")
    parser.add_argument("--out-dir", type=Path, required=True, help="Directory where JSON, manifest, and images are written.")
    parser.add_argument("--filename-prefix", default="fal-image", help="Base name prefix for output files.")
    parser.add_argument("--task-slug", default="fal-image-task", help="Stable task slug for tracking.")
    parser.add_argument("--num-images", type=int, default=None, help="Override num_images.")
    parser.add_argument("--aspect-ratio", default=None, help="Override aspect_ratio.")
    parser.add_argument("--resolution", default=None, help="Override resolution.")
    parser.add_argument("--image-size", default=None, help="Override image_size.")
    parser.add_argument("--background", default=None, help="Override background.")
    parser.add_argument("--output-format", default=None, help="Override output_format.")
    parser.add_argument("--quality", default=None, help="Override quality.")
    parser.add_argument("--seed", type=int, default=None, help="Override seed.")
    parser.add_argument("--sync-mode", choices=["true", "false"], default=None, help="Override sync_mode.")
    parser.add_argument("--extra-json", default=None, help="Extra JSON object merged into the model arguments.")
    parser.add_argument("--headers-json", default=None, help="Extra JSON object merged into fal request headers.")
    parser.add_argument("--poll-interval", type=float, default=10.0, help="Seconds between status polls.")
    parser.add_argument("--timeout", type=int, default=900, help="Maximum seconds to wait for completion.")
    parser.add_argument("--no-wait", action="store_true", help="Submit the job and stop before polling.")
    parser.add_argument("--no-download", action="store_true", help="Poll to completion but skip image download.")
    parser.add_argument("--no-store-io", action="store_true", help="Disable X-Fal-Store-IO.")
    parser.add_argument("--allow-fallback", action="store_true", help="Do not set x-app-fal-disable-fallback=true.")
    parser.add_argument("--dry-run", action="store_true", help="Write a resolved manifest without submitting the job.")
    parser.add_argument("--estimate-unit-quantity", type=float, default=None, help="Billing units passed to the pricing estimate endpoint. Defaults to num_images or 1.")
    return parser


def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
    return build_parser().parse_args(argv)


def _bool_from_cli(value: str | None) -> bool | None:
    if value is None:
        return None
    return value == "true"


def _prompt_text(args: argparse.Namespace) -> str:
    if bool(args.prompt) == bool(args.prompt_file):
        raise SystemExit("Use exactly one of --prompt or --prompt-file")
    if args.prompt_file is not None:
        return read_text(args.prompt_file)
    return str(args.prompt).strip()


def _resolve_preset(args: argparse.Namespace) -> dict[str, Any]:
    presets = load_presets()
    if args.model_alias is None and args.endpoint_id is None:
        raise SystemExit("Use --model-alias or --endpoint-id")
    if args.model_alias is not None:
        preset = presets.get(args.model_alias)
        if preset is None:
            known = ", ".join(sorted(presets))
            raise SystemExit(f"Unknown model alias: {args.model_alias}. Known aliases: {known}")
        return preset
    return {
        "provider": "fal",
        "family": "custom",
        "endpoint_id": args.endpoint_id,
        "task_type": "text-to-image",
        "download_keys": ["images[0].url"],
        "defaults": {},
    }


def _resolved_image_inputs(args: argparse.Namespace) -> list[str]:
    resolved: list[str] = []
    for path in args.image_file or []:
        resolved.append(data_uri_for_file(path))
    for url in args.image_url or []:
        resolved.append(str(url))
    return resolved


def _input_source_manifest(args: argparse.Namespace, preset: dict[str, Any]) -> dict[str, Any]:
    return {
        "image_files": [repo_relative(path) for path in (args.image_file or [])],
        "image_urls": list(args.image_url or []),
        "input_image_field": preset.get("input_image_field"),
    }


def _resolve_arguments(args: argparse.Namespace, preset: dict[str, Any], prompt_text: str) -> tuple[dict[str, Any], dict[str, Any]]:
    task_type = str(preset.get("task_type", "text-to-image"))
    resolved = dict(preset.get("defaults", {}))
    resolved["prompt"] = prompt_text

    image_inputs = _resolved_image_inputs(args)
    if task_type == "image-edit":
        if not image_inputs:
            raise SystemExit("Edit presets require at least one --image-file or --image-url")
        image_field = str(preset.get("input_image_field", "image_urls"))
        resolved[image_field] = image_inputs
    elif image_inputs and preset.get("input_image_field"):
        resolved[str(preset["input_image_field"])] = image_inputs

    overrides: dict[str, Any] = {}
    if args.num_images is not None:
        overrides["num_images"] = args.num_images
    if args.aspect_ratio is not None:
        overrides["aspect_ratio"] = args.aspect_ratio
    if args.resolution is not None:
        overrides["resolution"] = args.resolution
    if args.image_size is not None:
        overrides["image_size"] = args.image_size
    if args.background is not None:
        overrides["background"] = args.background
    if args.output_format is not None:
        overrides["output_format"] = args.output_format
    if args.quality is not None:
        overrides["quality"] = args.quality
    if args.seed is not None:
        overrides["seed"] = args.seed
    if args.sync_mode is not None:
        overrides["sync_mode"] = _bool_from_cli(args.sync_mode)

    resolved.update(overrides)
    resolved.update(coerce_json_object(args.extra_json))
    return resolved, overrides


def _estimate_cost(api_key: str, endpoint_id: str, unit_quantity: float) -> dict[str, Any] | None:
    payload = build_unit_price_estimate_payload(endpoint_id, unit_quantity)
    try:
        estimate = platform_post("/models/pricing/estimate", api_key, payload).payload
        if isinstance(estimate, dict):
            return estimate
    except SystemExit:
        pass

    pricing = platform_get("/models/pricing", api_key, {"endpoint_id": endpoint_id}).payload
    prices = pricing.get("prices") if isinstance(pricing, dict) else None
    if isinstance(prices, list) and prices:
        first = prices[0]
        if isinstance(first, dict):
            unit_price = first.get("unit_price")
            currency = first.get("currency")
            unit = first.get("unit")
            if isinstance(unit_price, (int, float)):
                return {
                    "estimate_type": "unit_price_fallback",
                    "total_cost": float(unit_price) * unit_quantity,
                    "currency": currency,
                    "unit": unit,
                    "unit_quantity": unit_quantity,
                }
    return None


def run_image_job(args: argparse.Namespace) -> dict[str, Any]:
    preset = _resolve_preset(args)
    prompt_text = _prompt_text(args)
    resolved_arguments, overrides = _resolve_arguments(args, preset, prompt_text)

    request_headers = default_platform_headers(
        store_io=not args.no_store_io,
        disable_fallback=not args.allow_fallback,
    )
    request_headers.update(coerce_json_object(args.headers_json))

    started_at = now_utc_iso()
    estimate_unit_quantity = args.estimate_unit_quantity
    if estimate_unit_quantity is None:
        value = resolved_arguments.get("num_images", 1)
        estimate_unit_quantity = float(value) if isinstance(value, (int, float, str)) else 1.0

    api_key = None if args.dry_run else require_fal_key()
    estimated_cost = (
        _estimate_cost(api_key, str(preset["endpoint_id"]), estimate_unit_quantity)  # type: ignore[arg-type]
        if api_key is not None
        else None
    )

    manifest: dict[str, Any] = {
        "timestamp": started_at,
        "task_slug": args.task_slug,
        "provider": preset.get("provider", "fal"),
        "model_alias": args.model_alias,
        "family": preset.get("family"),
        "endpoint_id": preset["endpoint_id"],
        "task_type": preset.get("task_type"),
        "status": "dry_run" if args.dry_run else "pending",
        "prompt_text": prompt_text,
        "prompt_hash": prompt_sha256(prompt_text),
        "input_source": _input_source_manifest(args, preset),
        "resolved_arguments": resolved_arguments,
        "preset_defaults": preset.get("defaults", {}),
        "explicit_overrides": overrides,
        "headers": request_headers,
        "request_id": None,
        "output_files": [],
        "output_urls": [],
        "estimated_cost": estimated_cost,
        "estimated_cost_method": estimated_cost.get("estimate_type") if isinstance(estimated_cost, dict) else None,
        "reconciled_cost": None,
        "cost_currency": estimated_cost.get("currency") if isinstance(estimated_cost, dict) else None,
        "raw_files": {},
        "notes": [],
    }

    args.out_dir.mkdir(parents=True, exist_ok=True)
    manifest_path = args.out_dir / f"{args.filename_prefix}-run.json"
    write_json(manifest_path, manifest)

    if args.dry_run:
        return manifest

    create_response = None
    create_payload: dict[str, Any]
    request_id: str
    response_url: str | None = None
    status_url: str | None = None

    try:
        import fal_client  # type: ignore

        os.environ.setdefault("FAL_KEY", api_key)
        client = fal_client.SyncClient(key=api_key)
        live_arguments = dict(resolved_arguments)
        image_field = preset.get("input_image_field")
        if image_field and args.image_file:
            encoded_inputs = [fal_client.encode_file(str(path)) for path in args.image_file]
            encoded_inputs.extend(args.image_url or [])
            live_arguments[str(image_field)] = encoded_inputs
        handle = client.submit(str(preset["endpoint_id"]), live_arguments, headers=request_headers)
        request_id = handle.request_id
        response_url = handle.response_url
        status_url = handle.status_url
        create_payload = {
            "request_id": handle.request_id,
            "response_url": handle.response_url,
            "status_url": handle.status_url,
            "cancel_url": handle.cancel_url,
        }
    except ImportError:
        create_response = queue_submit(str(preset["endpoint_id"]), api_key, resolved_arguments, headers=request_headers)
        create_payload = create_response.payload
        request_id = create_payload.get("request_id")
        if not isinstance(request_id, str) or not request_id:
            raise SystemExit("fal queue response did not include request_id")

    create_json_path = args.out_dir / f"{args.filename_prefix}-create.json"
    create_meta_path = args.out_dir / f"{args.filename_prefix}-create-meta.json"
    write_json(create_json_path, create_payload)
    if create_response is not None:
        write_json(create_meta_path, {"status_code": create_response.status_code, "headers": create_response.headers})
    else:
        write_json(create_meta_path, {"status_code": 200, "headers": {}, "client": "fal_client"})

    manifest["request_id"] = request_id
    manifest["status"] = "submitted"
    manifest["raw_files"]["create_json"] = repo_relative(create_json_path)
    manifest["raw_files"]["create_meta_json"] = repo_relative(create_meta_path)
    write_json(manifest_path, manifest)

    if args.no_wait:
        return manifest

    deadline = time.time() + args.timeout
    latest_status_headers: dict[str, Any] | None = None

    while True:
        if status_url is not None:
            status_response = queue_get_by_url(status_url, api_key, headers=request_headers, query={"logs": 1})
        else:
            status_response = queue_status(str(preset["endpoint_id"]), request_id, api_key, headers=request_headers, logs=True)
        latest_status_headers = status_response.headers
        status = str(status_response.payload.get("status", "")).upper()
        manifest["status"] = status.lower() if status else "unknown"
        write_json(args.out_dir / f"{args.filename_prefix}-status.json", status_response.payload)
        write_json(args.out_dir / f"{args.filename_prefix}-status-meta.json", {"status_code": status_response.status_code, "headers": status_response.headers})
        write_json(manifest_path, manifest)

        if status == "COMPLETED":
            break
        if time.time() >= deadline:
            raise SystemExit(f"Timed out waiting for request {request_id}")
        time.sleep(args.poll_interval)

    if response_url is not None:
        result_response = queue_get_by_url(response_url, api_key, headers=request_headers)
    else:
        result_response = queue_result(str(preset["endpoint_id"]), request_id, api_key, headers=request_headers)
    result_json_path = args.out_dir / f"{args.filename_prefix}-final.json"
    result_meta_path = args.out_dir / f"{args.filename_prefix}-final-meta.json"
    write_json(result_json_path, result_response.payload)
    write_json(result_meta_path, {"status_code": result_response.status_code, "headers": result_response.headers})

    media_outputs = collect_media_outputs(result_response.payload)
    downloaded_files: list[str] = []
    output_urls = [item["url"] for item in media_outputs]

    if not args.no_download:
        for index, item in enumerate(media_outputs, start=1):
            extension = Path(str(item.get("file_name", ""))).suffix or ".png"
            output_path = args.out_dir / f"{args.filename_prefix}-output-{index:02d}{extension}"
            download_file(str(item["url"]), output_path)
            downloaded_files.append(repo_relative(output_path))

    billable_units = None
    if latest_status_headers and "x-fal-billable-units" in latest_status_headers:
        billable_units = latest_status_headers["x-fal-billable-units"]
    if result_response.headers.get("x-fal-billable-units"):
        billable_units = result_response.headers["x-fal-billable-units"]

    manifest.update(
        {
            "status": "completed",
            "completed_at": now_utc_iso(),
            "output_files": downloaded_files,
            "output_urls": output_urls,
            "billable_units_header": billable_units,
            "request_headers_seen": {
                "status": latest_status_headers,
                "result": result_response.headers,
            },
            "raw_files": {
                **manifest["raw_files"],
                "status_json": repo_relative(args.out_dir / f"{args.filename_prefix}-status.json"),
                "status_meta_json": repo_relative(args.out_dir / f"{args.filename_prefix}-status-meta.json"),
                "final_json": repo_relative(result_json_path),
                "final_meta_json": repo_relative(result_meta_path),
            },
        }
    )

    if isinstance(result_response.payload, dict):
        result_error = result_response.payload.get("error") or result_response.payload.get("detail")
        if result_error:
            manifest["notes"].append(f"Result payload included error/detail: {result_error}")

    write_json(manifest_path, manifest)
    return manifest


def main() -> None:
    args = parse_args()
    run_image_job(args)


if __name__ == "__main__":
    main()