image-generation-gateways

ゲートウェイ経由の画像生成

ゲートウェイが標準化するのはアカウントアクセスであり、モデルの動作ではありません。ゲートウェイ、モデルのパブリッシャー、モデル／バージョン、実行経路、スキーマ、価格、安全性の動作、下流の利用規約は、それぞれ別の事実として、すべてのプランおよび台帳の行に記録してください。

このスキルでは、現在利用できる3つのマルチモデル対応サーバーレスゲートウェイを扱います。

• fal.ai Model APIs：モデルのエンドポイント ID に対する直接呼び出し、または永続キュー呼び出しに対応しており、キューのステータス確認、キャンセル、署名付き Webhook を利用できます。
• Replicate predictions：公式、コミュニティ、マーケットプレイス、またはバージョンを固定したモデルを対象とする予測リソースです。
• Together AI serverless images：パラメーターのサブセットがそれぞれ異なるモデルを対象とする、同期型の OpenAI 互換画像エンドポイントです。

これらのゲートウェイを選定した理由は、各社が公式に公開する契約上の仕様から、実質的に異なる3つの運用パターンが明らかになったためです。同じモデル名を提供しているというだけで、別のゲートウェイを追加しないでください。まず、そのコントロールプレーン、課金単位、データ経路、ライフサイクル、スキーマ、利用規約を調査してください。

ここに記載する変動しやすい情報は、2026-07-10 に確認しました。ユーザー向けの作業では、次の4つのエビデンスラベルを使用してください。

• FACT：引用したゲートウェイ、モデルのパブリッシャー、または適用される利用規約に直接記載されています。
• PROVIDER CLAIM：品質、商用利用、トレーニング、セキュリティ、同等性、またはインフラに関する表明ですが、ここでは独立した監査を行っていません。
• HEURISTIC：このスキルにおける保守的な本番運用の実践であり、ゲートウェイの制限ではありません。
• UNKNOWN：公開されている公式情報だけでは不十分です。アカウント、契約、DPA、モデルのパブリッシャー、またはサポートを通じて確認してください。

コードを書く前にルーティング記録を作成する

承認済みの記録に、次の項目を含めてください。

1. 本番環境での使用目的と画像の受け入れテスト。
2. ゲートウェイとアカウント／プロジェクト。
3. 正確なエンドポイント／モデル ID、パブリッシャー、公式／コミュニティ／パートナーの区分、バージョンポリシー。
4. スキーマ URL または API から取得したスキーマ、ならびに取得時刻とハッシュ。
5. すべての入力フィールド、参照元情報、出力数／形式／サイズ、安全性の制御、シード値。
6. 実行経路：同期、キュー／ポーリング、または Webhook。
7. 課金単位、単価のスナップショット、最大見積額、クレジット／割引、承認者。
8. 入力／出力の権利、モデルの利用規約、ゲートウェイの利用規約、下流の処理事業者／データ経路、保持期間、トレーニング設定、リージョン。
9. 成果物の保存先、バイト数／ピクセル数の上限、来歴フィールド、削除スケジュール。
10. 再試行、照合、キャンセル、フェイルオーバーのポリシー。

重要な項目に1つでも記載がない場合は、ドライランを準備し、送信する前に停止してください。

ロゴではなく、ワークロードに合わせて選ぶ

要件	fal.ai	Replicate	Together AI
長時間実行するサーバーレス処理の制御	キューがリクエスト／ステータス／レスポンス／キャンセル用 URL を返します。Webhook も利用できます	非同期予測がデフォルトです。ポーリング、キャンセル、または Webhook を利用できます	現在の /v1/images/generations は同期処理です。画像ジョブのリソースはドキュメントに記載されていません
バージョンの挙動	エンドポイント ID だけでは、モデルの不変のリビジョンであることを証明できません	コミュニティ予測では不変のバージョン ID を使用できます。公式モデルのエンドポイントは、保守されている最新バージョンに従います	モデル文字列は非推奨になったり、別のモデルにリダイレクトされたりする場合があります。元のバージョンを使う方法として、専用エンドポイントがドキュメントに記載されています
スキーマの確認	各モデルのページに個別のスキーマが公開されています。共通フィールドはプラットフォームのフィールドではありません	models.get で latest_version.openapi_schema を確認できます。バージョンのページには、バージョン固有のスキーマが保持されています	/v1/models でモデルのメタデータを確認できますが、画像パラメーターへの対応状況を確認するには、画像関連のドキュメントとモデル固有のページも必要です
出力先	モデルがインラインデータを返さない限り、公開 fal.media CDN の URL が使われます	現在のクライアントでは FileOutput が公開されます。URL は replicate.delivery を使用し、1 時間で有効期限が切れます	2 回目の URL 取得を避けるには、リクエストで response_format: base64 を指定します
データ制御	X-Fal-Store-IO: 0。CDN のライフサイクルは別途制御します	API の予測入力、出力、ログ、ファイルは、デフォルトで 1 時間後に削除されます	アカウントの ZDR／プライバシー設定を使用します。公開ドキュメント間でデフォルト設定に矛盾があるため、確認してください
隠れたルーティングのリスク	キューの再試行と、サポート対象モデルへのフォールバックがデフォルトで有効です	公式モデルが新しいバージョンに進む場合があります	非推奨ポリシーにより、選択されたモデル ID はリダイレクトされることがあります

どの項目も「最良」を意味するものではありません。固定した受け入れテストを使って、実際の要件でベンチマークを実施してください。ゲートウェイやモデルのビルドをまたいで同じシードを使っても、再現性を比較できるとは限りません。

各モデルの契約を固定する

prompt、width、steps、seed を含む汎用ペイロードを実装して、すべてのモデルに送信してはいけません。承認済みの各アダプターについて、次の作業を行います。

1. ファーストパーティのページまたは API から、最新のモデル固有スキーマを取得します。
2. そのスキーマに含まれるフィールドのみを許可し、ローカルで型と範囲を検証します。
3. 正規化したスキーマのハッシュとモデル／ページの URL をリリース記録に保存します。
4. ドライリクエストのフィクスチャと、モックによる成功／失敗レスポンスをテストします。
5. スキーマのハッシュ、エンドポイント ID、返されたモデル／バージョン、価格の単位、公開者の規約、安全性に関するデフォルト設定が変更された場合は、レビューを必須にします。

fal.ai の設計に影響する事実

• REST 認証には Authorization: Key <FAL_KEY> を使用します。
• エンドポイント fal-ai/flux/schnell の本番用キューは POST https://queue.fal.run/fal-ai/flux/schnell です。キューの状態は IN_QUEUE、IN_PROGRESS、COMPLETED で、完了後にレスポンス URL を取得します。
• キューリクエストは、選択されたサーバー／接続障害時に最大10回まで再試行できます。対応モデルでは、再試行後に fal が「同等」のフォールバックエンドポイントへルーティングする場合があります。これは同等性に関するプロバイダーの主張であり、モデル、重み、スキーマ、ライセンス、安全性、出力が同一であることの証明ではありません。
• 承認済みの1回の送信を、承認済みエンドポイントでの1回の実行に限定するには、X-Fal-No-Retry: 1 と x-app-fal-disable-fallback: 1 を使用します。クライアントのタイムアウトによって、サーバー側の処理が停止するわけではありません。
• X-Fal-Store-IO: 0 を指定すると、プラットフォームによる JSON 入出力履歴の保存を防げますが、CDN メディアの保存は防げません。JSON ペイロードのデフォルト保存期間は30日と記載されています。X-Fal-Object-Lifecycle-Preference は生成メディアの有効期限を制御します。設定されていない場合、アカウントのデフォルトは無期限の場合があります。
• fal のメディア URL は、URL を持っている人なら誰でもアクセスできます。fal CDN に入力としてアップロードされたファイルは、リクエストペイロードを削除しても削除されません。
• モデルの料金はエンドポイントごとに異なり、画像数、メガピクセル数、または計算秒数に基づく場合があります。GET https://api.fal.ai/v1/models/pricing?endpoint_id=... を照会し、アカウント固有の結果を使用してください。
• fal の Webhook は ED25519 を使用し、公開鍵は https://rest.fal.ai/.well-known/jwks.json から取得します。4つの X-Fal-Webhook-* ヘッダー、生のボディの SHA-256、ならびに文書化されている前後300秒のタイムスタンプ検証を使用します。

Replicate の設計に影響する事実

• REST 認証には Authorization: Bearer <REPLICATE_API_TOKEN> を使用します。
• バージョンを固定する汎用予測リクエストは、version と input を含む POST https://api.replicate.com/v1/predictions です。公式モデルの場合は、バージョンを指定せずに POST /v1/models/{owner}/{name}/predictions を使用します。Replicate が最新の状態を維持します。
• 予測の状態は starting、processing、succeeded、failed、canceled です。デフォルトでは非同期です。Prefer: wait=n は HTTP 呼び出しを1～60秒間だけ保留し、予測が未完了のまま返る場合があります。Cancel-After は5秒から24時間までのサーバー側の期限を設定します。
• API による予測の入力、出力、ファイル、ログは、デフォルトで1時間後に削除されます。Web から作成された予測データは、削除されるまで無期限に保持されます。
• 公式モデルは安定した API と出力ベースの料金を提供しますが、基盤となるバージョンは最新のものに維持されます。コミュニティ版はバージョンを固定できますが、サポートとスキーマは作成者が管理し、実行時間やハードウェアに基づく課金が一般的です。
• 現行の出力ファイルガイドでは、URL と認証の処理には SDK の FileOutput を使用し、replicate.delivery の URL には Authorization が不要とされています。一方、一般的な HTTP リファレンスでは、ファイル URL に Authorization が必要とされています。これはドキュメント間の不整合として扱ってください。現行の SDK FileOutput を優先するか、文書化されている replicate.delivery の URL のみを API トークンを転送せずに取得してください。任意のホストに認証情報を付けて401エラーを「修正」しないでください。
• Replicate の Webhook は、ユーザー／組織ごとの whsec_ シークレットの Base64 部分を使用し、{webhook-id}.{webhook-timestamp}.{raw-body} に対する HMAC-SHA256 を検証します。配信は重複したり、順不同で届いたりする場合があります。終端状態の配信は、完了からおよそ1分後まで、バックオフを挟んで再試行されます。
• 現行のデフォルト制限は、予測作成が毎分600回、その他の呼び出しが毎分3000回と記載されていますが、クレジットが少ない場合はより厳しい制限が適用されます。429エラーからの復旧には、最新の制御情報に従ってください。

設計に影響する Together AI の事実

• 現在のベース URL は https://api.together.ai/v1 です。Bearer 認証を使用してください。古いファーストパーティーのモデルページには api.together.xyz と記載されていますが、現在の API／互換性ドキュメントと SDK のデフォルトを使用してください。
• POST /v1/images/generations は画像レスポンスを同期的に返します。トランスポートのタイムアウトは、作成が完了したかどうか判別できない状態を意味します。公開されている冪等性キーや画像ジョブの照会機能は見つかりませんでした。
• このエンドポイントには共通フィールドがありますが、利用可否はモデルによって異なります。画像の概要では FLUX Schnell が aspect_ratio を使用すると説明されていますが、現在の FLUX Schnell のモデルページでは width／height の使用例が示され、エンドポイントのリファレンスにもこれらのフィールドが公開されています。これはファーストパーティーのドキュメント間の食い違いとして扱い、検証済みのスキーマ／SDK バージョンを固定し、対応していると推測しないでください。image_url と reference_images も、それぞれ異なるモデル群に適用されます。
• response_format は url または base64 です。n は 1～4 と記載されています。disable_safety_checker は false にしてください。一部のモデルではゲートウェイのセーフティチェッカーが実行されないため、アプリケーション側でのポリシー適用が引き続き必要です。
• サーバーレス画像生成の料金はモデルごとに異なり、通常は画像単位またはメガピクセル単位です。指定されたデフォルトを超えるステップ数には追加料金が発生する場合があります。動的なモデル別レート制限はレスポンスヘッダーで返されます。
• 非リダイレクトでの削除と有効なリダイレクトの両方が、非推奨ポリシーに記載されています。返された model と要求したモデルを比較し、一致しない場合は隔離してください。モデル文字列は常に不変のリビジョンを示すとは限りません。
• Together の Privacy and Security ページでは、デフォルトでは入力／出力を保存しないと説明されています。一方、現在の Privacy Policy と Terms では、保存／トレーニングの設定を「No」にすることで ZDR を有効にすると説明されています。デフォルトは UNKNOWN として扱い、プロジェクト設定と締結済みの契約を確認してください。ZDR は将来に適用されるものであり、遡及適用されません。

有料呼び出しの障壁

スキーマの読み取り、価格の読み取り、プラン、およびローカル／モックによる検証では、生成は許可されません。無料のエンドポイントや前払いクレジットでも、限られたリソースを消費します。

ライブ生成のたびに、正規の UUIDv4 試行 ID、同日中の価格／スキーマ／規約のスナップショット、レビュー済みの正の単価と見積もり、および有限の正の上限額が必要です。送信可能なアダプターを実行する前に、同梱の Python 3.11+ 標準ライブラリヘルパー scripts/validate_plan.py を使用して、オフライン承認プランを検証してください。

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>

このヘルパーは、プロバイダー SDK のインポート、認証情報の読み取り、ネットワーク呼び出し、メディアの生成、有料リクエストの許可、ジョブの送信を一切行いません。ゲートウェイ／モデル／バージョンのポリシー、スキーマ URL と SHA-256、正規ペイロードの形状を確認します。また、定義済みのすべての承認プランオブジェクト階層で未知のキーを拒否し、プラン全体をスキャンして秘密情報を含む可能性のあるキーや値を検出し、NaN や Infinity などの非標準 JSON 定数を拒否します。さらに、出力数とポリシー、認証情報を含まない HTTPS スキーマ／価格 URL、価格のソース／鮮度／通貨／単位／単価／課金対象ユニット数、10 進数演算による estimate == unit_price * billable_units の確認、ceiling >= estimate の確認、権利／モデレーション／ガバナンスのダイジェスト、および正規の UUIDv4 試行 ID を検証します。fal の価格照会で安全であると文書化されているクエリパラメーター endpoint_id は price.source で許可されますが、秘密情報を含むクエリパラメーターや、秘密情報らしいクエリ値は許可されません。このヘルパーは、安定した機密情報削除済み JSON を出力し、その出力内の文書化された承認エンベロープを対象に approval_sha256 を計算するため、受け入れられたすべての意味的フィールドがダイジェストに反映されます。終了コード 0 はプランが有効であること、2 は検証または期待ダイジェストとの不一致に失敗したこと、3 は解析または運用上の失敗を示します。

試行 ID、承認参照、正確なエンドポイント／モデル／ペイロード／件数、出力先とポリシー、スキーマハッシュ、価格ソース／単位／下限価格／見積もり／上限額、ガバナンス上の処置を含む、1 つの正規の承認記録をハッシュ化します。GATEWAY_APPROVAL_SHA256 はその記録と完全に一致していなければなりません。自由形式のチケット文字列はゲートにはなりません。

POST の前に O_CREAT|O_EXCL を使用して試行台帳を作成します。試行 UUID と承認ダイジェストは一度しか使用できません。プロバイダー ID は、ポーリング、出力検証、アセットのダウンロードを行う前に、直ちに永続化します。後続のプロセスが GATEWAY_RESUME=1 を使用できるのは、同じ永続化済み fal／Replicate ID の場合に限ります。Together のタイムアウト、または ID がない試行は、再送信せずに照合する必要があります。

探索的なバリアントを自動生成しないでください。プロバイダーが同一ジョブであると証明しない限り、再試行やゲートウェイ間のフェイルオーバーには、新たに承認された create が必要です。

価格観測の例。デフォルト値ではありません（2026-07-10）： fal は fal-ai/flux/schnell を $0.003/MP と記載しています。Replicate の公式 black-forest-labs/flux-schnell ページでは $3/1000 output images と示されています。Together は black-forest-labs/FLUX.1-schnell を $0.0027/MP、デフォルトのステップ数を 4 として掲載しています。アカウント割引、丸め、リクエストするサイズ／件数／ステップ数、モデルのリビジョン、税金によって、見積もりは変わることがあります。再度照会してください。

アダプターの完全な例

以下の Python 3.11+ の例は、1 つに変換されたペイロードではなく、意図的に 3 つのアダプターを使用しています。参照画像のアップロードポリシーを普遍的な仕様と誤解しないよう、テキストから画像への生成のみを使用します。デフォルトではオフラインプランを使用し、正確な承認と正の支出下限額を紐付け、Replicate のモデル／バージョンの OpenAPI を事前検証し、作成前に排他的な台帳を書き込み、fal の再試行／フォールバックを無効にし、既知の非同期 ID のみを再開し、Together には base64 を要求します。また、リダイレクトを拒否し、Pillow で画像バイト列を検証し、サニタイズ済みのリリース／課金／来歴マニフェストを書き込みます。

デコード検査用に Pillow をインストールします：python -m pip install Pillow。gateway_image.py としてコピーしてください。

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),
                "actual_cost_usd": None,
                "actual_cost_status": "provider billing line not returned; reconciliation required",
                "credits_discounts": os.getenv("GATEWAY_CREDITS_DISCOUNTS", "not reported"),
            },
            "lifecycle": {
                "retry_events": ledger.get("events", []),
                "provider_internal_retry": "disabled" if plan.gateway == "fal" else "not requested",
                "fallback": "disabled" if plan.gateway == "fal" else "none",
                "failover": "none",
                "cancel": "none",
                "unresolved_ambiguity": False,
            },
            "data_and_rights": {
                **governance,
                "data_control": {
                    "fal": "store-io=0; generated CDN expiry requested at 3600s",
                    "replicate": "API prediction default cleanup documented at 1h",
                    "together": "account ZDR setting separately verified by approval",
                }[plan.gateway],
            },
            "review": {
                "moderation_disposition": governance["moderation_disposition"],
                "human_review_disposition": governance["human_review_disposition"],
                "provenance_check": governance["provenance_result"],
            },
            "artifacts": artifacts,
            "source_urls_stored": False,
            "deletion_state": {
                "provider_source": "not retained in manifest; provider lifecycle applies",
                "local": "durable artifact retained under approved destination",
            },
        }
        manifest_path = destination / f"manifest-{attempt_id}.json"
        atomic_json(manifest_path, manifest)
        manifest_hash = hashlib.sha256(manifest_path.read_bytes()).hexdigest()
        completed_at = utc_now()
        update_ledger(
            ledger_path,
            state="complete",
            completed_at=completed_at,
            terminal_state="succeeded",
            manifest_file=manifest_path.name,
            manifest_sha256=manifest_hash,
            updated_at=completed_at,
        )
        print(json.dumps(manifest, indent=2))
    except AmbiguousCreate as exc:
        if mutable_ledger and ledger_path.exists():
            update_ledger(
                ledger_path,
                state="unresolved_ambiguous_create",
                unresolved_ambiguity=True,
                error_type=type(exc).__name__,
                error_at=utc_now(),
                updated_at=utc_now(),
            )
        raise
    except GatewayTerminalError as exc:
        if mutable_ledger and ledger_path.exists():
            update_ledger(
                ledger_path,
                state="terminal_failed",
                terminal_state="failed_or_canceled",
                error_type=type(exc).__name__,
                error_at=utc_now(),
                updated_at=utc_now(),
            )
        raise
    except GatewayHTTPError as exc:
        if mutable_ledger and ledger_path.exists():
            state = "needs_reconciliation" if provider_id else "create_rejected"
            update_ledger(
                ledger_path,
                state=state,
                error_type=type(exc).__name__,
                http_status=exc.code,
                error_at=utc_now(),
                updated_at=utc_now(),
            )
        raise
    except Exception as exc:
        if mutable_ledger and ledger_path.exists():
            state = "needs_reconciliation" if provider_id else "pre_submit_failed"
            update_ledger(
                ledger_path,
                state=state,
                error_type=type(exc).__name__,
                error_at=utc_now(),
                updated_at=utc_now(),
            )
        raise
    finally:
        try:
            os.unlink(lock_path)
        except OSError:
            pass


if __name__ == "__main__":
    main()

ドライランの例

GATEWAY_SEND が未設定のため、3つのコマンドはすべてオフラインで実行されます。まず、UUIDv4 の試行 ID と、レビュー済みのスキーマ、価格スナップショット、権利／利用条件スナップショットのハッシュを設定します。ドライランの概要には、正確な approval_sha256_required が出力されます。これが null の場合は、必須のレビュー項目が不足しています。

$env:IMAGE_PROMPT='Editorial product photograph of one unbranded cobalt desk lamp, three-quarter view, soft side light, neutral seamless background, no text'
$env:IMAGE_OUTPUT_DIR='C:\approved\gateway-output'
$env:GATEWAY_ATTEMPT_ID='<new canonical UUIDv4>'
$env:GATEWAY_APPROVAL_ID='change-ticket-or-human-approval-reference'
$env:GATEWAY_SCHEMA_SHA256='<64-hex reviewed schema hash>'
$env:GATEWAY_PRICE_SNAPSHOT_SHA256='<64-hex same-day account price snapshot hash>'
$env:GATEWAY_RIGHTS_TERMS_SHA256='<64-hex terms snapshot hash>'
$env:GATEWAY_DATA_CLASS='public'
$env:GATEWAY_RETENTION_REVIEW='approved for this provider path'
$env:GATEWAY_MODERATION_DISPOSITION='approved for generation'
$env:GATEWAY_HUMAN_REVIEW_DISPOSITION='required before release'
$env:GATEWAY_PROVENANCE_RESULT='not-applicable' # or passed / not-present-accepted

$env:IMG_GATEWAY='fal'
$env:GATEWAY_REVIEWED_UNIT_PRICE_USD='0.003'
$env:GATEWAY_ESTIMATED_USD='0.006' # square_hd is >1 MP and fal rounds up MP
$env:GATEWAY_MAX_USD='0.01'
python .\gateway_image.py

$env:IMG_GATEWAY='replicate'
# Optional strict version: obtain a reviewed 64-hex version and its schema from Replicate.
$env:REPLICATE_VERSION_ID=''
$env:GATEWAY_REVIEWED_UNIT_PRICE_USD='0.003'
$env:GATEWAY_ESTIMATED_USD='0.003'
$env:GATEWAY_MAX_USD='0.01'
python .\gateway_image.py

$env:IMG_GATEWAY='together'
$env:GATEWAY_REVIEWED_UNIT_PRICE_USD='0.0027'
$env:GATEWAY_ESTIMATED_USD='0.003' # 1024x1024 exceeds one decimal MP
$env:GATEWAY_MAX_USD='0.01'
python .\gateway_image.py

正確な概要をレビューし、その正確なレコードについて承認を得た後、紐付けられたフィールドを変更せずに、そのダイジェストをコピーします。

$env:IMG_GATEWAY='together'
$env:TOGETHER_API_KEY='<secret-store injection>'
$env:GATEWAY_APPROVAL_SHA256='<approval_sha256_required from the unchanged dry summary>'
$env:GATEWAY_SEND='1'
python .\gateway_image.py

承認済みの呼び出しの直後に、承認変数と送信変数の設定を解除してください。このコードは、プロンプト、トークン、生のレスポンス、base64、ソース URL をログに記録しません。非同期プロバイダー ID が永続化された後にプロセスが終了した場合は、同一の環境設定で GATEWAY_RESUME=1 を設定して再実行してください。Together では再開機能を使用しないでください。プロセスが突然終了した後に .lock が残っていても、意図的に自動削除はされません。オペレーターが削除する前に、記録された試行を照合してください。

Webhook コントロールプレーンは相互に置き換えられません

fal の場合：

• パースする前に生のボディを検証します。
• ドキュメントに記載されている X-Fal-Webhook-Request-Id、User-Id、Timestamp、Signature の4つのヘッダーを必須とします。
• ドキュメントに記載されている300秒の範囲を外れたタイムスタンプを拒否します。
• 生のボディをハッシュ化し、キャッシュ済みの ED25519 JWKS キーを使用して、改行区切りのメッセージを検証します。キーは24時間以内に更新します。
• リトライによって異なる値になることがあるため、request_id と gateway_request_id の両方で重複を排除します。
• Webhook の終端ステータスは OK／ERROR であり、キューの COMPLETED ではないことに注意してください。

Replicate の場合：

• /v1/webhooks/default/secret から組織／ユーザーの署名シークレットを取得し、安全にキャッシュします。
• HMAC-SHA256 を使用して、生の {id}.{timestamp}.{body} を検証します。スペース区切りのヘッダーに含まれる v1, 署名はすべて対象とし、定時間比較を行います。
• システム内で文書化したタイムスタンプの許容範囲を適用します（レビューしたページでは、Replicate は許容範囲を設けるよう求めていますが、具体的な数値は指定していません）。
• webhook-id と prediction ID の組み合わせで重複を排除します。順序が前後した中間イベントを受け入れ、終端状態の prediction を後退させたり再開したりしないでください。
• 部分的なログ／出力が本当に必要な場合を除き、webhook_events_filter: ["completed"] を指定します。

レビューした Together のリアルタイム画像エンドポイントには、Webhook によるライフサイクル管理がありません。同期待ち時間を偽の「ジョブ ID」で包まないでください。

リトライする前に照合する

レビュー済みの作成エンドポイントでは、クライアントの冪等性キーは公開されていません。POST ボディがゲートウェイに到達した可能性がある場合に、接続タイムアウトまたは読み取りタイムアウトが発生したときは、次の手順に従ってください。

1. 自動的に再送信しないでください。
2. ローカルの試行 ID、時間範囲、ゲートウェイ／モデル、ペイロードハッシュ、課金見積もりを保持します。
3. ゲートウェイのリクエスト ID または prediction ID を受信している場合は、そのリソースを正確に指定してポーリングします。
4. fal の使用状況明細、Replicate の predictions／課金情報、または Together のコスト分析／アカウントサポートを使用して照合します。
5. 試行を、成功、終端状態での失敗、キャンセル済み、または未解決に分類します。
6. 課金／出力の可能性が未解決の間に新たな作成を行う場合は、新しい承認を必須とします。

キャンセルは、実行されなかったことと同じではありません。Replicate のドキュメントによると、公式モデルのキャンセルでも課金される場合があり、時間課金の実行では経過した実行時間分が請求されます。fal にはキャンセル URL がありますが、選択したモデルの最新の課金ドキュメントを確認してください。Together の同期画像ルートには、一般公開されたキャンセル操作がありません。

フェイルオーバーではパラメーターではなく意図を維持する

本番環境で管理された運用を行う場合は、fal へのサイレントなフォールバックを無効にしてください。アプリケーション管理のセカンダリーゲートウェイを使用する場合は、次の点に従ってください。

• 2つ目のモデル、スキーマ、公開元／ライセンス、データ経路、価格上限、受け入れ基準を事前に承認します。
• プライマリーの作成処理が終了状態になるか、費用が確定した後にのみフェイルオーバーします。
• フィールド名を変更するのではなく、プロバイダー固有のペイロードを新たに作成します。
• シード、安全性の判断、アスペクト比プリセット、画像数、ネガティブプロンプト、参照画像の強度が引き継がれることを期待しないでください。
• 実際に使用したゲートウェイ／モデル／バージョンをアーティファクトに記録し、回帰テストのベースラインに黙って混在させないでください。
• セマンティック、ジオメトリ、テキスト、類似性、ポリシーの各チェックを再実行します。

視覚的な完全再現性が必要な場合、フェイルオーバーは利用できません。固定されたモデルビルド／デプロイメントと、検証済みの決定論的な設定を使用してください。

入力とアーティファクトの管理

公開 URL があるという理由だけで参照画像を送信しないでください。著作権／ライセンス、プライバシー／肖像権に関する同意、生体認証／子ども／規制対象に該当するかどうか、許可された変換内容を確認してください。ソースハッシュと権利記録を保管してください。

ゲートウェイ固有の入力に関する注意事項：

• fal SDK によるアップロードでは、公開 CDN URL が生成されます。ペイロード削除 API では入力 CDN ファイルは削除されません。適切なプライベートなエンタープライズ経路が契約上確認されていない限り、機密性の高い参照画像は使用しないでください。
• Replicate クライアントは、ホストされた URL、100 MB までのローカルファイル、または1 MB 未満の場合にのみ推奨されるデータ URI を受け付けます。API の予測データは通常1時間後に削除されますが、マーケットプレイス／サードパーティーの利用規約も適用されます。
• Together の画像編集では、モデルによって image_url または reference_images を使用します。公開ドキュメントでは、これらの画像ルートに共通するプライベートなアップロード経路は確認できません。

すべての出力について、直ちに管理された耐久性のあるコピーを作成してください。URL 出力には HTTPS／プロバイダーのオリジンを必須とし、リダイレクトを禁止します。アセットホストにゲートウェイの認証情報を送信しないでください。バイト数／時間の上限、MIME タイプとマジックナンバー、完全なデコード、ピクセル数／フレーム数の上限、アトミックな書き込み、暗号学的ハッシュを適用してください。埋め込みのプロベナンスが削除されないよう、生のバイト列を保持してください。モデルまたはリリースポリシーで必要な場合は、C2PA／透かしのメタデータを別途検査してください。ゲートウェイ URL はプロベナンスではありません。

権利とデータガバナンスの境界

モデルを利用できるからといって、権利が付与されるわけではありません。ゲートウェイの規約と、実行時点で有効だったモデル提供元の正確な規約を記録してください。

• falの事実： 各モデルには独自のライセンスがあり、商用利用可または研究利用限定と表示される場合があります。パートナーAPIモデルでは、クライアントのコンテンツがサードパーティプロバイダーに転送される場合があります。falの「Enterprise Ready」バッジはプロバイダーによる分類であり、法的助言ではありません。
• Replicateの事実： 現行の規約では、出力に関する権利はサードパーティ規約に従い、BFL、Ideogram、Stabilityなどに固有の制限も含まれています。コミュニティモデルは作成者によって保守されています。公式のFLUX Schnellページでは現在、商用利用可および「Zero training」がうたわれていますが、バッジはモデルページ上の表示として扱い、組み込まれている規約を確認してください。
• Togetherの事実： 現行の規約では、規約間に矛盾がある場合、サードパーティモデルの規約が適用されます。FLUX SchnellのページにはApache 2.0のウェイトと記載されていますが、出力および入力に関する義務については、引き続き現行のBFLおよびTogetherの規約を確認する必要があります。Togetherは、モデルがTogetherのインフラストラクチャ上で実行されるため、サードパーティモデルの作成者がリクエストを受け取ることはないと説明しています。これはプロバイダーによる説明として扱い、パートナー固有の例外がないことを確認してください。

著作権保護の対象となること、独占性、非侵害、補償、商標やパブリシティ権のクリアランスを保証しないでください。安全フィルターは、利用許可でも法的審査でもありません。

2026-07-10時点で確認したデータの取り扱い：

• falは、JSONペイロードのデフォルト保存期間が30日であること、X-Fal-Store-IO: 0を任意で指定できること、公開CDNの保持期間は別途制御されること、ベンダー／サービスプロバイダーが関与すること、チームがリクエストを閲覧できることを文書化しています。顧客入力データのトレーニング利用、正確な処理リージョン、パートナーAPIの保持期間は、該当する契約／モデルの経路が不明な限り不明です。
• Replicateは、API予測データが1時間後に削除されること、および米国に所在するサブプロセッサーを文書化しています。規約では、サービスの提供に伴う処理と、その結果得られるデータの利用が認められています。マーケットプレイスのモデルでは、別個の処理者や規約が適用される場合があります。モデルページの「Zero training」は、プラットフォーム全体に対する保証ではありません。
• Togetherのプライバシーポリシーでは、トレーニング利用はオプトインであるとされ、ZDRについて説明されています。プライバシー／セキュリティ関連の文書では、入力／出力はデフォルトで保存されないとされています。ただし、有効化に関する記述が矛盾しているため、デフォルト設定を推測せず、設定と契約を確認してください。エンタープライズ向けのデータレジデンシー／プライベートネットワーキングは個別の取り決めで利用できますが、全体に適用されるリージョンは不明です。

個人情報、機密情報、規制対象のコンテンツ、またはデータレジデンシーの要件があるコンテンツについては、締結済みのDPA／セキュリティ審査と、正確なルーティングの確認を必須としてください。ゲートウェイの一般的なポリシーだけでは、下流のプロプライエタリモデルへの経路について判断できません。

リリースと課金の台帳

台帳の行に以下が記録されるまで、実行を完了として扱わないでください。

• ローカルの試行ID、ゲートウェイのリクエスト／予測／レスポンスID、タイムスタンプ、終端状態。
• リクエストしたゲートウェイ／エンドポイント／モデル／バージョン、および返されたゲートウェイ／モデル／バージョン。
• スキーマのURL／ハッシュ／日付、クライアント／SDKのバージョン。
• ペイロードのハッシュ、シード、件数、リクエストした／実際の寸法と形式。
• 料金の情報源／時刻、単位、推定最大額、利用可能な場合は実際の使用明細、クレジット／割引、承認ID。
• 再試行／フォールバック／フェイルオーバー／キャンセルのイベント、および未解決の曖昧さ。
• 保持期間／ZDR／store-IOの設定、実行／データ経路の分類、権利／規約のスナップショット。
• モデレーションと人によるレビューの判定。
• 永続アーティファクトのハッシュ、バイトサイズ、デコードメタデータ、来歴の確認、削除状態。

リクエストしたIDと返されたIDの不一致、スキーマ／価格の変更、既知のリクエストIDがないのに課金された場合、成功したリクエストにアーティファクトがない場合、課金／出力の重複、または承認を受けていないアーティファクトについて、アラートを設定してください。

一次情報の参照元

2026-07-10時点で確認：

fal.ai

• モデル API の概要
• 非同期キュー
• 信頼性、再試行、フォールバック
• プラットフォームヘッダーと保持制御
• Webhook の検証
• 料金と料金 API
• FLUX Schnell のスキーマ
• FAQ：公開メディアとモデル固有のライセンス
• API サービス利用規約、利用規約、プライバシー

Replicate

• HTTP 予測 API とステータス
• 公式モデルとコミュニティモデル
• モデル／バージョンのスキーマ検出
• 予測の作成と期限
• 入力ファイル、出力ファイル、保持
• Webhook の配信と署名の検証
• 請求、料金、レート制限
• FLUX Schnell モデル／API ページ、利用規約、プライバシー、サブプロセッサー

Together AI

• 画像エンドポイントのリファレンス
• 画像の概要、パラメーター、参照画像
• 現在のサーバーレスモデルカタログとモデルメタデータ API
• モデルのライフサイクル／非推奨化
• 動的レート制限、請求分析、料金
• プライバシーとセキュリティ、プライバシーポリシー、利用規約
• FLUX Schnell モデルページ

情報源に食い違いがある場合、都合のよい記述を黙って選ばないでください。食い違いを記録し、より限定的で安全な動作を採用したうえで、実運用や機密性の高い作業を始める前に、アカウントまたは契約上の確認を得てください。