Image Generation Gateways

python scripts/validate_plan.py approval-plan.json --max-age-days 1
python scripts/validate_plan.py approval-plan.json --max-age-days 1 --expected-approval-sha256 <64-hex-digest>

from __future__ import annotations

import base64
import binascii
import hashlib
import ipaddress
import json
import math
import os
import platform
import random
import socket
import tempfile
import time
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from email.utils import parsedate_to_datetime
from io import BytesIO
from pathlib import Path
from typing import Any, Callable
from urllib.error import HTTPError, URLError
from urllib.parse import urlsplit
from urllib.request import HTTPRedirectHandler, Request, build_opener

JSON_CAP = 48 * 1024 * 1024
IMAGE_CAP = 48 * 1024 * 1024
PIXEL_CAP = 36_000_000
POLL_DEADLINE_S = 360
CREATE_DEADLINE_S = 90
ASSET_DEADLINE_S = 120
READ_SLICE_S = 5
SCHEMA_CHECKED = "2026-07-10"
PRICE_CHECKED = "2026-07-10"
OUTPUT_POLICY_VERSION = "image-v1"


class StopRedirects(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        raise RuntimeError(f"redirect refused: {code}")


HTTP = build_opener(StopRedirects)


class GatewayHTTPError(RuntimeError):
    def __init__(self, code: int, safe_message: str):
        super().__init__(safe_message)
        self.code = code


class GatewayTerminalError(RuntimeError):
    pass


class AmbiguousCreate(RuntimeError):
    pass


@dataclass(frozen=True)
class Plan:
    gateway: str
    model: str
    url: str
    payload: dict[str, Any]
    lifecycle: str
    version_policy: str
    schema_url: str
    price_source: str
    price_unit: str
    known_unit_price: Decimal
    billable_units: Decimal


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


def canonical(value: Any) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode()


def sha256_value(value: Any) -> str:
    return hashlib.sha256(canonical(value)).hexdigest()


def require_sha256(name: str, value: str) -> str:
    value = value.strip().lower()
    if len(value) != 64 or any(char not in "0123456789abcdef" for char in value):
        raise ValueError(f"{name} must be a 64-character lowercase hex SHA-256")
    return value


def parse_positive_money(name: str, value: str, required: bool) -> Decimal | None:
    value = value.strip()
    if not value:
        if required:
            raise ValueError(f"{name} is required")
        return None
    try:
        amount = Decimal(value)
    except InvalidOperation as exc:
        raise ValueError(f"{name} must be a finite positive decimal") from exc
    if not amount.is_finite() or amount <= 0:
        raise ValueError(f"{name} must be a finite positive decimal")
    return amount


def valid_attempt_id(value: str) -> str:
    try:
        parsed = uuid.UUID(value)
    except ValueError as exc:
        raise ValueError("GATEWAY_ATTEMPT_ID must be a UUIDv4") from exc
    normalized = str(parsed)
    if parsed.version != 4 or value.lower() != normalized:
        raise ValueError("GATEWAY_ATTEMPT_ID must be a canonical UUIDv4")
    return normalized


def valid_provider_id(value: Any, provider: str) -> str:
    if not isinstance(value, str) or not value or len(value) > 256:
        raise AmbiguousCreate(f"{provider} create returned no usable identity")
    if any(char not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-" for char in value):
        raise AmbiguousCreate(f"{provider} create returned an unsafe identity")
    return value


def public_https(url: str, allowed: tuple[str, ...]) -> str:
    parsed = urlsplit(url)
    host = (parsed.hostname or "").lower()
    if parsed.scheme != "https" or not host or parsed.username or parsed.password or parsed.fragment:
        raise ValueError("asset must be credential-free fragment-free HTTPS")
    if parsed.port not in (None, 443):
        raise ValueError("asset port is not allowed")
    if not any(host == suffix or host.endswith("." + suffix) for suffix in allowed):
        raise ValueError("asset host is not provider-approved")
    for answer in socket.getaddrinfo(host, 443, type=socket.SOCK_STREAM):
        if not ipaddress.ip_address(answer[4][0]).is_global:
            raise ValueError("asset resolved to a non-public address")
    return url


def api_url(url: str, host: str, prefix: str) -> str:
    parsed = urlsplit(url)
    expected_path = parsed.path == prefix or parsed.path.startswith(prefix.rstrip("/") + "/")
    if (
        parsed.scheme != "https"
        or parsed.hostname != host
        or parsed.port not in (None, 443)
        or parsed.username
        or parsed.password
        or parsed.fragment
        or not expected_path
    ):
        raise ValueError("unexpected API URL")
    return url


def retry_after_seconds(headers: Any, now: datetime | None = None) -> float | None:
    value = headers.get("Retry-After") if headers is not None else None
    if not value:
        return None
    try:
        seconds = float(value)
        return max(0.0, min(seconds, 60.0)) if math.isfinite(seconds) else None
    except ValueError:
        try:
            target = parsedate_to_datetime(value)
            if target.tzinfo is None:
                target = target.replace(tzinfo=timezone.utc)
            current = now or datetime.now(timezone.utc)
            return max(0.0, min((target - current).total_seconds(), 60.0))
        except (TypeError, ValueError, OverflowError):
            return None


def bounded_read(response: Any, cap: int, deadline: float, expected_type: str | None) -> bytes:
    if time.monotonic() >= deadline:
        raise TimeoutError("response deadline reached")
    length = response.headers.get("Content-Length")
    if length:
        try:
            declared = int(length)
        except ValueError as exc:
            raise RuntimeError("invalid Content-Length") from exc
        if declared < 0 or declared > cap:
            raise RuntimeError("response exceeds client cap")
    if expected_type and response.headers.get_content_type() != expected_type:
        raise RuntimeError("unexpected response content type")
    chunks: list[bytes] = []
    size = 0
    reader = getattr(response, "read1", response.read)
    while True:
        if time.monotonic() >= deadline:
            raise TimeoutError("response deadline reached")
        chunk = reader(min(64 * 1024, cap + 1 - size))
        if not chunk:
            break
        size += len(chunk)
        if size > cap:
            raise RuntimeError("response exceeds client cap")
        chunks.append(chunk)
    return b"".join(chunks)


def safe_error_hash(exc: HTTPError, deadline: float) -> str:
    try:
        raw = bounded_read(exc, 16_384, deadline, None)
    except Exception:
        raw = b""
    finally:
        try:
            exc.close()
        except OSError:
            pass
    return hashlib.sha256(raw).hexdigest()


def post_json(
    url: str,
    expected_host: str,
    path_prefix: str,
    auth_header: str,
    payload: dict[str, Any],
    extra_headers: dict[str, str] | None = None,
) -> dict[str, Any]:
    api_url(url, expected_host, path_prefix)
    body = json.dumps(payload, separators=(",", ":")).encode()
    headers = {
        "Accept": "application/json",
        "Authorization": auth_header,
        "Content-Type": "application/json",
    }
    headers.update(extra_headers or {})
    deadline = time.monotonic() + CREATE_DEADLINE_S
    try:
        timeout = max(0.1, min(READ_SLICE_S, deadline - time.monotonic()))
        with HTTP.open(Request(url, data=body, headers=headers, method="POST"), timeout=timeout) as response:
            raw = bounded_read(response, JSON_CAP, deadline, "application/json")
        value = json.loads(raw)
        if not isinstance(value, dict):
            raise RuntimeError("gateway create response is not an object")
        return value
    except HTTPError as exc:
        error_hash = safe_error_hash(exc, deadline)
        wait = retry_after_seconds(exc.headers)
        suffix = f"; retry_after={wait:g}s" if wait is not None else ""
        if exc.code >= 500:
            raise AmbiguousCreate(
                f"create returned HTTP {exc.code}; body_sha256={error_hash}; reconcile before resubmission"
            ) from exc
        raise GatewayHTTPError(
            exc.code, f"gateway rejected create with HTTP {exc.code}; body_sha256={error_hash}{suffix}"
        ) from None
    except GatewayHTTPError:
        raise
    except AmbiguousCreate:
        raise
    except Exception as exc:
        raise AmbiguousCreate(
            "create transport or 2xx response failed before a durable identity; do not resubmit"
        ) from exc


def sleep_with_deadline(seconds: float, deadline: float) -> None:
    remaining = deadline - time.monotonic()
    if remaining <= 0:
        raise TimeoutError("gateway deadline reached")
    time.sleep(min(max(0.0, seconds), remaining))


def get_json(
    url: str,
    expected_host: str,
    path_prefix: str,
    auth_header: str,
    deadline: float,
    event: Callable[[dict[str, Any]], None] | None = None,
) -> dict[str, Any]:
    api_url(url, expected_host, path_prefix)
    delay = 1.0
    while True:
        if time.monotonic() >= deadline:
            raise TimeoutError("read-only gateway deadline reached")
        headers = {"Accept": "application/json", "Authorization": auth_header}
        try:
            timeout = max(0.1, min(READ_SLICE_S, deadline - time.monotonic()))
            with HTTP.open(Request(url, headers=headers, method="GET"), timeout=timeout) as response:
                raw = bounded_read(response, JSON_CAP, deadline, "application/json")
            value = json.loads(raw)
            if not isinstance(value, dict):
                raise RuntimeError("gateway response is not an object")
            return value
        except HTTPError as exc:
            retryable = exc.code == 429 or exc.code >= 500
            error_hash = safe_error_hash(exc, deadline)
            if not retryable:
                raise GatewayHTTPError(
                    exc.code, f"gateway read failed with HTTP {exc.code}; body_sha256={error_hash}"
                ) from None
            wait = retry_after_seconds(exc.headers)
            pause = wait if wait is not None else delay + random.uniform(0, delay * 0.15)
            if event:
                event({"kind": "read_retry", "http_status": exc.code, "wait_seconds": pause})
        except (TimeoutError, URLError, RuntimeError, json.JSONDecodeError) as exc:
            if time.monotonic() >= deadline:
                raise TimeoutError("read-only gateway deadline reached") from exc
            pause = delay + random.uniform(0, delay * 0.15)
            if event:
                event({"kind": "read_retry", "error": type(exc).__name__, "wait_seconds": pause})
        sleep_with_deadline(pause, deadline)
        delay = min(delay * 1.7, 12.0)


def check_image(blob: bytes) -> tuple[str, str, int, int]:
    if not blob or len(blob) > IMAGE_CAP:
        raise ValueError("image is empty or exceeds client cap")
    if blob.startswith(b"\x89PNG\r\n\x1a\n"):
        kind, suffix = "image/png", ".png"
    elif blob.startswith(b"\xff\xd8\xff"):
        kind, suffix = "image/jpeg", ".jpg"
    elif blob.startswith(b"RIFF") and blob[8:12] == b"WEBP":
        kind, suffix = "image/webp", ".webp"
    else:
        raise ValueError("output magic is not PNG, JPEG, or WebP")
    try:
        from PIL import Image, UnidentifiedImageError

        Image.MAX_IMAGE_PIXELS = PIXEL_CAP
        with Image.open(BytesIO(blob)) as image:
            width, height = image.size
            if width * height > PIXEL_CAP or getattr(image, "n_frames", 1) != 1:
                raise ValueError("output exceeds pixel/frame policy")
            image.verify()
        with Image.open(BytesIO(blob)) as image:
            image.load()
    except ImportError as exc:
        raise RuntimeError("Pillow is required for artifact validation") from exc
    except (UnidentifiedImageError, OSError) as exc:
        raise ValueError("output failed full image decode") from exc
    return kind, suffix, width, height


def fetch_asset(url: str, domains: tuple[str, ...], deadline: float) -> bytes:
    public_https(url, domains)
    request = Request(
        url,
        headers={"Accept": "image/png,image/jpeg,image/webp", "User-Agent": "gateway-image-example/1"},
        method="GET",
    )
    if time.monotonic() >= deadline:
        raise TimeoutError("asset deadline reached")
    timeout = max(0.1, min(READ_SLICE_S, deadline - time.monotonic()))
    try:
        with HTTP.open(request, timeout=timeout) as response:
            content_type = response.headers.get_content_type()
            if content_type not in {"image/png", "image/jpeg", "image/webp"}:
                raise RuntimeError("asset response is not an approved image MIME")
            blob = bounded_read(response, IMAGE_CAP, deadline, None)
    except HTTPError as exc:
        error_hash = safe_error_hash(exc, deadline)
        raise GatewayHTTPError(
            exc.code, f"asset fetch failed with HTTP {exc.code}; body_sha256={error_hash}"
        ) from None
    mime, _, _, _ = check_image(blob)
    if content_type != mime:
        raise RuntimeError("asset MIME/magic mismatch")
    return blob


def atomic_bytes(path: Path, blob: bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    handle: int | None = None
    temp_name: str | None = None
    try:
        handle, temp_name = tempfile.mkstemp(prefix=path.name + ".", dir=path.parent)
        try:
            stream = os.fdopen(handle, "wb")
            handle = None
            with stream:
                stream.write(blob)
                stream.flush()
                os.fsync(stream.fileno())
        except Exception:
            if handle is not None:
                try:
                    os.close(handle)
                except OSError:
                    pass
                handle = None
            raise
        os.replace(temp_name, path)
        temp_name = None
    finally:
        if handle is not None:
            try:
                os.close(handle)
            except OSError:
                pass
        if temp_name and os.path.exists(temp_name):
            try:
                os.unlink(temp_name)
            except OSError:
                pass


def atomic_json(path: Path, value: dict[str, Any]) -> None:
    atomic_bytes(path, (json.dumps(value, indent=2, sort_keys=True) + "\n").encode())


def exclusive_json(path: Path, value: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
    if hasattr(os, "O_BINARY"):
        flags |= os.O_BINARY
    handle: int | None = None
    created = False
    try:
        handle = os.open(path, flags, 0o600)
        created = True
        try:
            stream = os.fdopen(handle, "wb")
            handle = None
            with stream:
                stream.write((json.dumps(value, indent=2, sort_keys=True) + "\n").encode())
                stream.flush()
                os.fsync(stream.fileno())
        except Exception:
            if handle is not None:
                try:
                    os.close(handle)
                except OSError:
                    pass
                handle = None
            raise
    except Exception:
        if created:
            try:
                os.unlink(path)
            except OSError:
                pass
        raise
    finally:
        if handle is not None:
            try:
                os.close(handle)
            except OSError:
                pass


def read_local_json(path: Path) -> dict[str, Any]:
    raw = path.read_bytes()
    if len(raw) > JSON_CAP:
        raise RuntimeError("local ledger exceeds cap")
    value = json.loads(raw)
    if not isinstance(value, dict):
        raise RuntimeError("local ledger is not an object")
    return value


def update_ledger(path: Path, **changes: Any) -> dict[str, Any]:
    current = read_local_json(path)
    current.update(changes)
    atomic_json(path, current)
    return current


def append_ledger_event(path: Path, event: dict[str, Any]) -> None:
    current = read_local_json(path)
    events = current.get("events")
    if not isinstance(events, list):
        events = []
    if len(events) >= 100:
        events = events[-99:]
    events.append({"at": utc_now(), **event})
    current["events"] = events
    current["updated_at"] = utc_now()
    atomic_json(path, current)


def make_plan() -> Plan:
    gateway = os.getenv("IMG_GATEWAY", "fal").lower()
    prompt = os.getenv("IMAGE_PROMPT", "").strip()
    if not prompt:
        raise ValueError("IMAGE_PROMPT is required")
    seed = int(os.getenv("IMAGE_SEED", "1"))
    if gateway == "fal":
        model = "fal-ai/flux/schnell"
        return Plan(
            gateway,
            model,
            "https://queue.fal.run/fal-ai/flux/schnell",
            {
                "prompt": prompt,
                "image_size": "square_hd",
                "num_inference_steps": 4,
                "num_images": 1,
                "seed": seed,
                "enable_safety_checker": True,
                "output_format": "png",
            },
            "queue",
            "floating endpoint ID; schema snapshot required",
            "https://fal.ai/docs/model-api-reference/image-generation-api/flux-schnell",
            "https://api.fal.ai/v1/models/pricing?endpoint_id=fal-ai/flux/schnell",
            "megapixel-rounded-up",
            Decimal("0.003"),
            Decimal("2"),
        )
    if gateway == "replicate":
        model = "black-forest-labs/flux-schnell"
        version = os.getenv("REPLICATE_VERSION_ID", "").strip().lower()
        inputs = {
            "prompt": prompt,
            "aspect_ratio": "1:1",
            "num_outputs": 1,
            "num_inference_steps": 4,
            "seed": seed,
            "output_format": "webp",
            "output_quality": 90,
            "disable_safety_checker": False,
            "go_fast": False,
            "megapixels": "1",
        }
        if version:
            require_sha256("REPLICATE_VERSION_ID", version)
            return Plan(
                gateway,
                model,
                "https://api.replicate.com/v1/predictions",
                {"version": version, "input": inputs},
                "prediction",
                "immutable version requested and preflight-required",
                f"https://api.replicate.com/v1/models/black-forest-labs/flux-schnell/versions/{version}",
                "https://replicate.com/black-forest-labs/flux-schnell/api",
                "output-image",
                Decimal("0.003"),
                Decimal("1"),
            )
        return Plan(
            gateway,
            model,
            "https://api.replicate.com/v1/models/black-forest-labs/flux-schnell/predictions",
            {"input": inputs},
            "prediction",
            "official model follows maintained latest; schema preflight-required",
            "https://api.replicate.com/v1/models/black-forest-labs/flux-schnell",
            "https://replicate.com/black-forest-labs/flux-schnell/api",
            "output-image",
            Decimal("0.003"),
            Decimal("1"),
        )
    if gateway == "together":
        model = "black-forest-labs/FLUX.1-schnell"
        return Plan(
            gateway,
            model,
            "https://api.together.ai/v1/images/generations",
            {
                "model": model,
                "prompt": prompt,
                "width": 1024,
                "height": 1024,
                "steps": 4,
                "n": 1,
                "seed": seed,
                "response_format": "base64",
                "output_format": "png",
                "disable_safety_checker": False,
            },
            "synchronous",
            "model ID may redirect; response identity enforced",
            "https://docs.together.ai/reference/post-images-generations",
            "https://www.together.ai/pricing",
            "megapixel",
            Decimal("0.0027"),
            Decimal("1.048576"),
        )
    raise ValueError("IMG_GATEWAY must be fal, replicate, or together")


def merge_schema(schema: dict[str, Any]) -> dict[str, Any]:
    if "allOf" not in schema:
        return schema
    merged: dict[str, Any] = {"properties": {}, "required": []}
    for part in schema.get("allOf", []):
        if isinstance(part, dict):
            part = merge_schema(part)
            merged["properties"].update(part.get("properties", {}))
            merged["required"].extend(part.get("required", []))
    return merged


def replicate_input_schema(openapi: dict[str, Any]) -> dict[str, Any]:
    schemas = openapi.get("components", {}).get("schemas", {})
    candidate = schemas.get("Input") if isinstance(schemas, dict) else None
    if not isinstance(candidate, dict):
        raise RuntimeError("Replicate OpenAPI has no Input schema")
    candidate = merge_schema(candidate)
    if not isinstance(candidate.get("properties"), dict):
        raise RuntimeError("Replicate Input schema has no properties")
    return candidate


def validate_schema_value(name: str, value: Any, schema: dict[str, Any]) -> None:
    expected = schema.get("type")
    valid = {
        "string": isinstance(value, str),
        "integer": isinstance(value, int) and not isinstance(value, bool),
        "number": isinstance(value, (int, float)) and not isinstance(value, bool),
        "boolean": isinstance(value, bool),
        "array": isinstance(value, list),
        "object": isinstance(value, dict),
    }
    if expected in valid and not valid[expected]:
        raise RuntimeError(f"Replicate input {name!r} violates schema type")
    if "enum" in schema and value not in schema["enum"]:
        raise RuntimeError(f"Replicate input {name!r} violates schema enum")
    if isinstance(value, (int, float)) and not isinstance(value, bool):
        if "minimum" in schema and value < schema["minimum"]:
            raise RuntimeError(f"Replicate input {name!r} is below schema minimum")
        if "maximum" in schema and value > schema["maximum"]:
            raise RuntimeError(f"Replicate input {name!r} exceeds schema maximum")


def validate_replicate_inputs(openapi: dict[str, Any], inputs: dict[str, Any]) -> None:
    schema = replicate_input_schema(openapi)
    properties = schema["properties"]
    unknown = sorted(set(inputs) - set(properties))
    missing = sorted(set(schema.get("required", [])) - set(inputs))
    if unknown or missing:
        raise RuntimeError(f"Replicate schema mismatch; unknown={unknown}; missing={missing}")
    for name, value in inputs.items():
        field = properties[name]
        if not isinstance(field, dict):
            raise RuntimeError(f"Replicate input schema for {name!r} is invalid")
        validate_schema_value(name, value, field)


def preflight_replicate(plan: Plan, token: str, expected_hash: str) -> dict[str, Any]:
    deadline = time.monotonic() + CREATE_DEADLINE_S
    auth = "Bearer " + token
    info = get_json(plan.schema_url, "api.replicate.com", "/v1/models/black-forest-labs/flux-schnell", auth, deadline)
    requested_version = plan.payload.get("version")
    if requested_version:
        if info.get("id") != requested_version:
            raise RuntimeError("Replicate version preflight did not return the requested version")
        version = requested_version
        openapi = info.get("openapi_schema")
    else:
        if info.get("owner") not in (None, "black-forest-labs") or info.get("name") not in (
            None,
            "flux-schnell",
        ):
            raise RuntimeError("Replicate model preflight returned a different model")
        latest = info.get("latest_version")
        if not isinstance(latest, dict) or not isinstance(latest.get("id"), str):
            raise RuntimeError("Replicate model preflight has no latest version")
        version = latest["id"]
        openapi = latest.get("openapi_schema")
    if not isinstance(openapi, dict):
        raise RuntimeError("Replicate preflight has no OpenAPI schema")
    actual_hash = hashlib.sha256(canonical(openapi)).hexdigest()
    if actual_hash != expected_hash:
        raise RuntimeError("Replicate schema hash drifted; do not create")
    validate_replicate_inputs(openapi, plan.payload["input"])
    return {
        "url": plan.schema_url,
        "sha256": actual_hash,
        "checked_at": utc_now(),
        "version": version,
        "validation": "model/version identity and Input allowlist passed before create",
    }


def wait_fal(
    plan: Plan,
    token: str,
    on_identity: Callable[[str, dict[str, Any]], None],
    on_state: Callable[[str], None],
    on_retry: Callable[[dict[str, Any]], None],
    resume_id: str | None = None,
) -> tuple[str, dict[str, Any], list[bytes]]:
    auth = "Key " + token
    deadline = time.monotonic() + POLL_DEADLINE_S
    if resume_id:
        request_id = valid_provider_id(resume_id, "fal")
    else:
        initial = post_json(
            plan.url,
            "queue.fal.run",
            "/fal-ai/flux/schnell",
            auth,
            plan.payload,
            {
                "X-Fal-Store-IO": "0",
                "X-Fal-No-Retry": "1",
                "x-app-fal-disable-fallback": "1",
                "X-Fal-Object-Lifecycle-Preference": json.dumps(
                    {"expiration_duration_seconds": 3600}
                ),
            },
        )
        request_id = valid_provider_id(initial.get("request_id"), "fal")
        on_identity(request_id, initial)
    root = f"https://queue.fal.run/{plan.model}/requests/{request_id}"
    while True:
        state = get_json(
            root + "/status",
            "queue.fal.run",
            f"/{plan.model}/requests",
            auth,
            deadline,
            on_retry,
        )
        if state.get("request_id") not in (None, request_id):
            raise RuntimeError("fal request_id mismatch")
        status = state.get("status")
        on_state(str(status))
        if status == "COMPLETED":
            if state.get("error"):
                raise GatewayTerminalError(f"fal completed with an error; id={request_id}")
            result = get_json(
                root + "/response",
                "queue.fal.run",
                f"/{plan.model}/requests",
                auth,
                deadline,
                on_retry,
            )
            images = result.get("images")
            if not isinstance(images, list) or len(images) != 1 or not isinstance(images[0], dict):
                raise RuntimeError("fal output does not match frozen schema")
            url = images[0].get("url")
            if not isinstance(url, str):
                raise RuntimeError("fal image URL missing")
            asset_deadline = min(deadline, time.monotonic() + ASSET_DEADLINE_S)
            return request_id, result, [fetch_asset(url, ("fal.media",), asset_deadline)]
        if status not in {"IN_QUEUE", "IN_PROGRESS"}:
            raise GatewayTerminalError(f"unexpected fal queue state: {status!r}; id={request_id}")
        sleep_with_deadline(1.0, deadline)


def wait_replicate(
    plan: Plan,
    token: str,
    expected_version: str,
    on_identity: Callable[[str, dict[str, Any]], None],
    on_state: Callable[[str], None],
    on_retry: Callable[[dict[str, Any]], None],
    resume_id: str | None = None,
) -> tuple[str, dict[str, Any], list[bytes]]:
    auth = "Bearer " + token
    deadline = time.monotonic() + POLL_DEADLINE_S
    if resume_id:
        prediction_id = valid_provider_id(resume_id, "Replicate")
        current = get_json(
            f"https://api.replicate.com/v1/predictions/{prediction_id}",
            "api.replicate.com",
            "/v1/predictions",
            auth,
            deadline,
            on_retry,
        )
    else:
        current = post_json(
            plan.url,
            "api.replicate.com",
            "/v1",
            auth,
            plan.payload,
            {"Cancel-After": "2m"},
        )
        prediction_id = valid_provider_id(current.get("id"), "Replicate")
        on_identity(prediction_id, current)
    while True:
        if current.get("id") != prediction_id:
            raise RuntimeError("Replicate prediction id mismatch")
        status = current.get("status")
        on_state(str(status))
        if status == "succeeded":
            break
        if status in {"failed", "canceled"}:
            raise GatewayTerminalError(f"Replicate prediction {status}; id={prediction_id}")
        if status not in {"starting", "processing"}:
            raise GatewayTerminalError(f"unexpected Replicate state: {status!r}; id={prediction_id}")
        sleep_with_deadline(1.0, deadline)
        current = get_json(
            f"https://api.replicate.com/v1/predictions/{prediction_id}",
            "api.replicate.com",
            "/v1/predictions",
            auth,
            deadline,
            on_retry,
        )
    if current.get("version") != expected_version:
        raise RuntimeError("Replicate returned a version different from the preflighted version")
    if current.get("model") not in (None, plan.model):
        raise RuntimeError("Replicate returned a different model ID")
    output = current.get("output")
    if not isinstance(output, list) or len(output) != 1 or not isinstance(output[0], str):
        raise RuntimeError("Replicate output does not match frozen URI-array schema")
    asset_deadline = min(deadline, time.monotonic() + ASSET_DEADLINE_S)
    return prediction_id, current, [
        fetch_asset(output[0], ("replicate.delivery",), asset_deadline)
    ]


def run_together(
    plan: Plan,
    token: str,
    on_identity: Callable[[str, dict[str, Any]], None],
    on_state: Callable[[str], None],
) -> tuple[str, dict[str, Any], list[bytes]]:
    response = post_json(
        plan.url,
        "api.together.ai",
        "/v1/images/generations",
        "Bearer " + token,
        plan.payload,
    )
    request_id = valid_provider_id(response.get("id"), "Together")
    on_identity(request_id, response)
    if response.get("model") != plan.model:
        raise RuntimeError("Together returned a different model ID; quarantine this run")
    data = response.get("data")
    if not isinstance(data, list) or len(data) != 1 or not isinstance(data[0], dict):
        raise RuntimeError("Together output does not match frozen schema")
    encoded = data[0].get("b64_json")
    encoded_cap = (IMAGE_CAP * 4 // 3) + 8
    if not isinstance(encoded, str) or len(encoded) > encoded_cap:
        raise RuntimeError("Together base64 output missing or oversized")
    try:
        blob = base64.b64decode(encoded, validate=True)
    except (ValueError, binascii.Error) as exc:
        raise RuntimeError("Together returned invalid base64") from exc
    check_image(blob)
    on_state("succeeded")
    return request_id, response, [blob]


def safe_metrics(response: dict[str, Any]) -> dict[str, Any] | None:
    candidate = response.get("metrics")
    if not isinstance(candidate, dict):
        candidate = response.get("timings")
    if not isinstance(candidate, dict):
        return None
    safe: dict[str, Any] = {}
    for key, value in candidate.items():
        if isinstance(key, str) and len(key) <= 64 and isinstance(value, (int, float, str, bool)):
            safe[key] = value
    return safe


def client_version() -> str:
    try:
        import PIL

        pillow = PIL.__version__
    except ImportError:
        pillow = "missing"
    return f"Python {platform.python_version()}; urllib stdlib; Pillow {pillow}"


def approval_record(
    plan: Plan,
    attempt_id: str,
    approval_id: str,
    schema_hash: str,
    price_snapshot_hash: str,
    reviewed_unit_price: Decimal,
    estimate: Decimal,
    maximum: Decimal,
    destination: Path,
    governance: dict[str, str],
) -> dict[str, Any]:
    output_policy = {
        "version": OUTPUT_POLICY_VERSION,
        "destination_sha256": hashlib.sha256(str(destination).encode()).hexdigest(),
        "artifact_naming": f"image-{attempt_id}-{{index}}.<decoded-extension>",
        "manifest_naming": f"manifest-{attempt_id}.json",
        "allowed_formats": ["png", "jpeg", "webp"],
        "max_bytes_each": IMAGE_CAP,
        "max_pixels_each": PIXEL_CAP,
        "max_frames_each": 1,
        "source_urls_stored": False,
        "atomic_same_directory": True,
    }
    return {
        "attempt_id": attempt_id,
        "approval_id_sha256": hashlib.sha256(approval_id.encode()).hexdigest(),
        "gateway": plan.gateway,
        "endpoint": plan.url,
        "model": plan.model,
        "version_policy": plan.version_policy,
        "payload_sha256": hashlib.sha256(canonical(plan.payload)).hexdigest(),
        "prompt_sha256": hashlib.sha256(os.environ["IMAGE_PROMPT"].encode()).hexdigest(),
        "count": 1,
        "schema": {
            "url": plan.schema_url,
            "sha256": schema_hash,
            "checked": SCHEMA_CHECKED,
        },
        "pricing": {
            "source": plan.price_source,
            "snapshot_sha256": price_snapshot_hash,
            "checked": PRICE_CHECKED,
            "unit": plan.price_unit,
            "reviewed_unit_price_usd": str(reviewed_unit_price),
            "billable_units": str(plan.billable_units),
            "known_public_unit_floor_usd": str(plan.known_unit_price),
            "estimated_usd": str(estimate),
            "max_usd": str(maximum),
        },
        "output_policy": output_policy,
        "governance": governance,
    }


def main() -> None:
    plan = make_plan()
    live = os.getenv("GATEWAY_SEND") == "1"
    resume_requested = os.getenv("GATEWAY_RESUME") == "1"
    estimate = parse_positive_money(
        "GATEWAY_ESTIMATED_USD", os.getenv("GATEWAY_ESTIMATED_USD", ""), live
    )
    maximum = parse_positive_money("GATEWAY_MAX_USD", os.getenv("GATEWAY_MAX_USD", ""), live)
    reviewed_unit_price = parse_positive_money(
        "GATEWAY_REVIEWED_UNIT_PRICE_USD",
        os.getenv("GATEWAY_REVIEWED_UNIT_PRICE_USD", ""),
        live,
    )
    if reviewed_unit_price is not None and reviewed_unit_price < plan.known_unit_price:
        raise ValueError("reviewed unit price is below the dated known public floor")
    if estimate is not None and reviewed_unit_price is not None:
        reviewed_floor = reviewed_unit_price * plan.billable_units
        if estimate < reviewed_floor:
            raise ValueError(f"GATEWAY_ESTIMATED_USD is below reviewed plan floor {reviewed_floor}")
    if estimate is not None and maximum is not None and maximum < estimate:
        raise ValueError("GATEWAY_MAX_USD is below the reviewed estimate")

    attempt_text = os.getenv("GATEWAY_ATTEMPT_ID", "").strip()
    approval_id = os.getenv("GATEWAY_APPROVAL_ID", "").strip()
    schema_text = os.getenv("GATEWAY_SCHEMA_SHA256", "").strip()
    price_snapshot_text = os.getenv("GATEWAY_PRICE_SNAPSHOT_SHA256", "").strip()
    rights_text = os.getenv("GATEWAY_RIGHTS_TERMS_SHA256", "").strip()
    governance = {
        "data_class": os.getenv("GATEWAY_DATA_CLASS", "").strip(),
        "retention_review": os.getenv("GATEWAY_RETENTION_REVIEW", "").strip(),
        "moderation_disposition": os.getenv("GATEWAY_MODERATION_DISPOSITION", "").strip(),
        "human_review_disposition": os.getenv("GATEWAY_HUMAN_REVIEW_DISPOSITION", "").strip(),
        "provenance_result": os.getenv("GATEWAY_PROVENANCE_RESULT", "").strip(),
        "rights_terms_sha256": rights_text.lower(),
    }
    destination = Path(os.getenv("IMAGE_OUTPUT_DIR", "gateway-output")).resolve()

    required_hash: str | None = None
    record: dict[str, Any] | None = None
    ready = all(
        [
            attempt_text,
            approval_id,
            schema_text,
            price_snapshot_text,
            rights_text,
            estimate is not None,
            maximum is not None,
            reviewed_unit_price is not None,
            *governance.values(),
        ]
    )
    if ready:
        attempt_id = valid_attempt_id(attempt_text)
        schema_hash = require_sha256("GATEWAY_SCHEMA_SHA256", schema_text)
        price_snapshot_hash = require_sha256(
            "GATEWAY_PRICE_SNAPSHOT_SHA256", price_snapshot_text
        )
        governance["rights_terms_sha256"] = require_sha256(
            "GATEWAY_RIGHTS_TERMS_SHA256", rights_text
        )
        if governance["provenance_result"] not in {
            "passed",
            "not-present-accepted",
            "not-applicable",
        }:
            raise ValueError("GATEWAY_PROVENANCE_RESULT is not an approved disposition")
        record = approval_record(
            plan,
            attempt_id,
            approval_id,
            schema_hash,
            price_snapshot_hash,
            reviewed_unit_price,
            estimate,
            maximum,
            destination,
            governance,
        )
        required_hash = hashlib.sha256(canonical(record)).hexdigest()

    summary = {
        "dry_run": not live,
        "gateway": plan.gateway,
        "model": plan.model,
        "lifecycle": plan.lifecycle,
        "version_policy": plan.version_policy,
        "schema_checked": SCHEMA_CHECKED,
        "price_checked": PRICE_CHECKED,
        "known_public_unit_floor_usd": str(plan.known_unit_price),
        "billable_units": str(plan.billable_units),
        "count": 1,
        "estimated_usd": str(estimate) if estimate is not None else None,
        "max_usd": str(maximum) if maximum is not None else None,
        "suggested_attempt_id": None if attempt_text else str(uuid.uuid4()),
        "approval_sha256_required": required_hash,
        "approval_record_ready": ready,
        "safety": "gateway control enabled where schema supports it",
    }
    print(json.dumps(summary, indent=2))
    if not live:
        print("OFFLINE PLAN ONLY")
        return
    if not ready or record is None or required_hash is None:
        raise RuntimeError("live generation requires every approval, schema, price, and governance field")
    if not resume_requested and datetime.now(timezone.utc).date().isoformat() != PRICE_CHECKED:
        raise RuntimeError("dated price floor is stale; refresh constants and approval")
    supplied_hash = require_sha256(
        "GATEWAY_APPROVAL_SHA256", os.getenv("GATEWAY_APPROVAL_SHA256", "")
    )
    if supplied_hash != required_hash:
        raise RuntimeError("approval hash does not exactly match this call and output policy")

    key_name = {
        "fal": "FAL_KEY",
        "replicate": "REPLICATE_API_TOKEN",
        "together": "TOGETHER_API_KEY",
    }[plan.gateway]
    token = os.getenv(key_name, "")
    if not token:
        raise RuntimeError(f"{key_name} is required only after exact approval")

    attempt_id = record["attempt_id"]
    state_dir = destination / ".gateway-attempts"
    ledger_path = state_dir / f"{attempt_id}.json"
    lock_path = state_dir / f"{attempt_id}.lock"
    try:
        exclusive_json(
            lock_path,
            {
                "attempt_id": attempt_id,
                "approval_sha256": supplied_hash,
                "pid": os.getpid(),
                "acquired_at": utc_now(),
                "policy": "stale lock requires operator reconciliation; never auto-break",
            },
        )
    except FileExistsError as exc:
        raise RuntimeError("attempt is already active or has a stale crash lock; reconcile first") from exc

    provider_id: str | None = None
    mutable_ledger = False
    schema_meta: dict[str, Any]
    try:
        if ledger_path.exists():
            if not resume_requested:
                raise RuntimeError("attempt ledger already exists; replay refused")
            ledger = read_local_json(ledger_path)
            if (
                ledger.get("approval_sha256") != supplied_hash
                or ledger.get("payload_sha256") != record["payload_sha256"]
                or ledger.get("gateway") != plan.gateway
            ):
                raise RuntimeError("resume does not match the durable attempt")
            if ledger.get("state") in {"complete", "terminal_failed", "create_rejected"}:
                raise RuntimeError("terminal attempt cannot be replayed")
            provider_id = ledger.get("provider_id")
            if not isinstance(provider_id, str) or plan.gateway == "together":
                raise RuntimeError("this unresolved attempt has no safe public resume path")
            schema_meta = ledger["schema"]
            mutable_ledger = True
            append_ledger_event(ledger_path, {"kind": "resume_started", "provider_id": provider_id})
        else:
            if resume_requested:
                raise RuntimeError("resume requested but no durable attempt ledger exists")
            if plan.gateway == "replicate":
                schema_meta = preflight_replicate(plan, token, record["schema"]["sha256"])
            else:
                schema_meta = {
                    "url": plan.schema_url,
                    "sha256": record["schema"]["sha256"],
                    "checked_at": utc_now(),
                    "version": None,
                    "validation": "reviewed frozen adapter hash supplied and approval-bound",
                }
            created_at = utc_now()
            exclusive_json(
                ledger_path,
                {
                    "attempt_id": attempt_id,
                    "state": "authorized_unsubmitted",
                    "created_at": created_at,
                    "updated_at": created_at,
                    "gateway": plan.gateway,
                    "endpoint": plan.url,
                    "requested_model": plan.model,
                    "version_policy": plan.version_policy,
                    "payload_sha256": record["payload_sha256"],
                    "prompt_sha256": record["prompt_sha256"],
                    "approval_id_sha256": record["approval_id_sha256"],
                    "approval_sha256": supplied_hash,
                    "approval_record_sha256": sha256_value(record),
                    "schema": schema_meta,
                    "pricing": record["pricing"],
                    "output_policy": record["output_policy"],
                    "governance": governance,
                    "provider_id": None,
                    "provider_status": None,
                    "events": [{"at": created_at, "kind": "ledger_created_before_submit"}],
                },
            )
            mutable_ledger = True

        def on_identity(identity: str, response: dict[str, Any]) -> None:
            nonlocal provider_id
            provider_id = identity
            returned = response.get("version") if plan.gateway == "replicate" else response.get("model")
            now = utc_now()
            update_ledger(
                ledger_path,
                state="submitted",
                provider_id=identity,
                returned_model_or_version=returned,
                submitted_at=now,
                identity_persisted_at=now,
                updated_at=now,
            )

        def on_state(status: str) -> None:
            update_ledger(
                ledger_path,
                state="polling" if status not in {"succeeded", "COMPLETED"} else "provider_succeeded",
                provider_status=status,
                last_state_at=utc_now(),
                updated_at=utc_now(),
            )

        def on_retry(event: dict[str, Any]) -> None:
            append_ledger_event(ledger_path, event)

        if plan.gateway == "fal":
            request_id, response, blobs = wait_fal(
                plan, token, on_identity, on_state, on_retry, provider_id
            )
        elif plan.gateway == "replicate":
            request_id, response, blobs = wait_replicate(
                plan,
                token,
                schema_meta["version"],
                on_identity,
                on_state,
                on_retry,
                provider_id,
            )
        else:
            if provider_id is not None:
                raise RuntimeError("Together synchronous creates cannot be resumed")
            request_id, response, blobs = run_together(plan, token, on_identity, on_state)

        terminal_at = utc_now()
        update_ledger(
            ledger_path,
            state="provider_succeeded",
            terminal_state="succeeded",
            terminal_at=terminal_at,
            provider_status="succeeded",
            updated_at=terminal_at,
        )
        artifacts: list[dict[str, Any]] = []
        for index, blob in enumerate(blobs):
            mime, suffix, width, height = check_image(blob)
            path = destination / f"image-{attempt_id}-{index}{suffix}"
            atomic_bytes(path, blob)
            artifacts.append(
                {
                    "file": path.name,
                    "sha256": hashlib.sha256(blob).hexdigest(),
                    "bytes": len(blob),
                    "mime": mime,
                    "width": width,
                    "height": height,
                    "frames": 1,
                    "decode": "Pillow verify plus full load passed",
                }
            )
        artifact_at = utc_now()
        update_ledger(
            ledger_path,
            state="artifact_durable",
            artifact_at=artifact_at,
            artifacts=artifacts,
            updated_at=artifact_at,
        )
        ledger = read_local_json(ledger_path)
        returned_identity = (
            response.get("version") if plan.gateway == "replicate" else response.get("model")
        )
        manifest = {
            "attempt_id": attempt_id,
            "gateway": plan.gateway,
            "request_id": request_id,
            "requested": {
                "endpoint": plan.url,
                "model": plan.model,
                "version_policy": plan.version_policy,
                "payload_sha256": record["payload_sha256"],
                "prompt_sha256": record["prompt_sha256"],
                "seed": plan.payload.get("seed", plan.payload.get("input", {}).get("seed")),
                "count": 1,
            },
            "returned_model_or_version": returned_identity,
            "terminal_state": "succeeded",
            "timestamps": {
                "ledger_created": ledger.get("created_at"),
                "submitted": ledger.get("submitted_at"),
                "identity_persisted": ledger.get("identity_persisted_at"),
                "terminal": terminal_at,
                "artifact_durable": artifact_at,
                "manifest_durable": utc_now(),
            },
            "schema": {
                **schema_meta,
                "client_version": client_version(),
            },
            "approval": {
                "approval_id_sha256": record["approval_id_sha256"],
                "approval_sha256": supplied_hash,
                "output_policy": record["output_policy"],
            },
            "pricing": {
                **record["pricing"],
                "actual_usage": safe_metrics(response),