image-generation-gateways

通过网关生成图像

网关统一的是账户访问方式，而非模型行为。在每份方案和每条账本记录中，都应将网关、模型发布方、模型/版本、执行路径、架构、价格、安全行为和下游条款分别记录。

本技能涵盖当前的三个多模型无服务器网关：

• fal.ai Model APIs：直接调用或通过持久队列调用模型端点 ID，并支持队列状态查询、取消和签名 Webhook。
• Replicate predictions：围绕官方、社区、市场或固定版本模型创建的预测资源。
• Together AI serverless images：同步的 OpenAI 兼容图像端点，涵盖参数子集各不相同的模型。

之所以选择这些网关，是因为其一手协议体现了三种实质不同的运行模式。不要仅仅因为另一个网关提供相同的模型名称，就将其加入。应先研究其控制平面、计费单位、数据路径、生命周期、架构和条款。

此处易变信息的核查日期为 2026-07-10。面向用户的内容应使用以下四种证据标签：

• FACT：引用的网关、模型发布方或管辖条款直接记载的事实。
• PROVIDER CLAIM：关于质量、商业用途、训练、安全、等效性或基础设施的陈述，本文未进行独立审计。
• HEURISTIC：本技能建议的保守生产实践，并非网关限制。
• UNKNOWN：公开的一手证据不足；请通过账户、合同、DPA、模型发布方或支持渠道确认。

编写代码前先建立路由记录

创建一份经签字确认的记录，包含：

1. 生产用途和图像验收测试；
2. 网关及账户/项目；
3. 确切的端点/模型 ID、发布方、官方/社区/合作伙伴状态，以及版本策略；
4. 架构 URL 或通过 API 获取的架构，以及检索时间和哈希值；
5. 完整输入字段、参考素材来源、输出数量/格式/尺寸、安全控制和随机种子；
6. 执行路径：同步、队列/轮询或 Webhook；
7. 计费单位、单价快照、预计最高费用、额度/折扣和审批人；
8. 输入/输出权利、模型条款、网关条款、下游处理方/数据路径、保留期限、训练设置和区域；
9. 产物存储位置、字节/像素限制、溯源字段和删除计划；
10. 重试、对账、取消和故障转移策略。

如果缺少任何重要信息，请先准备试运行，并在提交前停止。

根据工作负载选择，而非看 Logo

要求	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，一小时后过期	请求中设置 response_format: base64，以避免再次获取 URL
数据控制	使用 X-Fal-Store-IO: 0；CDN 生命周期另行控制	默认情况下，API 预测的输入、输出、日志和文件会在一小时后删除	账户的 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 可能会将请求路由到“等效”的备用端点。这是关于等效性的提供方声明，并不能证明模型、权重、架构、许可、安全性或输出完全相同。
• 如果一次获批的提交必须只执行一次获批的端点，请使用 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 的人都可以访问媒体。即使请求载荷已删除，上传到 fal CDN 的输入文件也不会被移除。
• 模型按端点计价，计价单位可能是每张图片、每百万像素或每计算秒。请查询 GET https://api.fal.ai/v1/models/pricing?endpoint_id=...，并使用适用于特定账户的结果。
• fal webhook 使用 ED25519，公钥来自 https://rest.fal.ai/.well-known/jwks.json；包含四个 X-Fal-Webhook-* 标头、原始请求体的 SHA-256，以及文档规定的 ±300 秒时间戳校验。

会影响设计的 Replicate 事实

• REST 身份验证格式为 Authorization: Bearer <REPLICATE_API_TOKEN>。
• 通用的固定版本预测：向 POST https://api.replicate.com/v1/predictions 发送请求，并提供 version 和 input。官方模型：向 POST /v1/models/{owner}/{name}/predictions 发送请求，无需指定版本；Replicate 会保持其为当前版本。
• 预测状态包括 starting、processing、succeeded、failed 和 canceled。默认采用异步方式。Prefer: wait=n 只会让 HTTP 调用等待 1-60 秒，返回的预测仍可能尚未完成。Cancel-After 会设置服务器端截止时间，范围为 5 秒至 24 小时。
• API 预测的输入、输出、文件和日志默认会在 1 小时后删除。通过网页创建的预测数据会无限期保留，直到被删除。
• 官方模型具有稳定的 API，按输出计价，但其底层版本会保持为最新版本。社区版本可以固定，但支持和架构由创建者控制，且通常按运行时间／硬件计费。
• 当前输出文件指南指出，SDK 的 FileOutput 应取代 URL／身份验证处理方式，并且 replicate.delivery URL 无需身份验证逻辑。通用 HTTP 参考文档仍指出文件 URL 需要身份验证。请将此视为文档冲突：优先使用当前的 SDK FileOutput，或仅获取文档说明的 replicate.delivery URL，且不要转发 API 令牌；切勿通过向任意主机附加凭据来“修复” 401 错误。
• Replicate webhook 使用 HMAC-SHA256，对 {webhook-id}.{webhook-timestamp}.{raw-body} 进行签名，使用每个用户／组织的 whsec_ 密钥中的 base64 部分。投递可能重复且顺序可能颠倒；终态投递会按退避策略重试，直到完成后约 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。部分模型不会运行网关安全检查器，因此仍需制定应用层策略。
• Serverless 图像定价因模型而异，通常按图像或百万像素计费；steps 超过列出的默认值可能会增加费用。响应标头会返回动态的、因模型而异的速率限制。
• 弃用政策既说明了不重定向的移除，也说明了生效中的重定向。请将返回的 model 与请求的模型进行比较，若不匹配则隔离处理。模型字符串并不总是不可变的修订版本。
• Together 的隐私与安全页面称默认不存储输入／输出，而当前隐私政策和服务条款则说明，可通过选择存储／训练选项的“No”来启用 ZDR。默认设置应视为未知；请核实项目设置和已签署的合同。ZDR 仅对未来数据生效，不具有追溯效力。

付费调用屏障

读取架构、读取价格、制定计划和本地／模拟验证都不授权生成。免费端点和预付额度仍会消耗有限资源。

每次实时生成前，都必须准备规范的 UUIDv4 尝试标识、当天的价格／架构／条款快照、经过审核且为正数的单价和估算金额，以及有限的正数上限。在任何具备提交能力的适配器运行前，使用随附的 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、价格来源／时效／货币／单位／单价／计费单位、使用十进制运算计算的 estimate == unit_price * billable_units、ceiling >= estimate、权利／审核／治理摘要，以及规范的 UUIDv4 尝试标识。文档中安全的 fal 价格查询参数 endpoint_id 可用于 price.source；包含密钥的查询参数和看起来像密钥的查询值则不允许。脚本会输出稳定且已脱敏的 JSON，并根据该输出中有文档说明的审批封装计算 approval_sha256，因此每个已接受的语义字段都会影响摘要。退出代码 0 表示计划有效，2 表示验证失败或预期摘要不匹配，3 表示解析或操作失败。

将尝试标识、审批引用、精确的端点／模型／载荷／数量、输出目标和策略、架构哈希、价格来源／单位／底价／估算金额／上限，以及治理处置记录在一起，生成一个规范授权记录的哈希。GATEWAY_APPROVAL_SHA256 必须与该记录完全一致；自由格式的工单字符串不能作为授权凭据。

在 POST 前，使用 O_CREAT|O_EXCL 创建尝试账本。尝试 UUID 和审批摘要均只能使用一次。轮询、输出验证或下载资源前，立即保存服务提供商 ID。后续进程只能在使用同一持久化 fal／Replicate 标识的情况下设置 GATEWAY_RESUME=1；Together 超时或没有 ID 的尝试必须先核对状态，不得重新提交。

绝不要自动生成探索性变体。重试或切换到其他网关，都属于新的已审批创建请求，除非服务提供商能证明它是同一个任务。

价格观察示例，并非默认值（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+ 示例刻意实现了三个适配器，而不是将载荷转换后用于所有适配器。它仅使用文生图，因此不会让人误以为参考图上传策略是通用约定。示例默认使用离线计划，绑定精确审批和正数支出底线，预检 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，这三条命令均为离线操作。首先设置一个 UUIDv4 尝试标识，以及已审查的架构、价格快照和权利/条款快照的哈希值。试运行摘要会打印确切的 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 和预测 ID 进行去重，接受乱序的中间状态，且绝不回退或重新打开终态预测；
• 除非确实需要部分日志/输出，否则请求 webhook_events_filter: ["completed"]。

Together 已审查的实时图像端点没有 Webhook 生命周期。不要把同步超时伪装成“作业 ID”。

重试前先核对

已审查的创建端点均未提供客户端幂等键。如果在 POST 请求体可能已到达网关后发生连接/读取超时：

1. 不要自动重新提交；
2. 保留本地尝试 ID、时间窗口、网关/模型、载荷哈希和费用估算；
3. 如果已收到网关请求/预测 ID，请轮询该确切资源；
4. 使用 fal 用量明细、Replicate 预测/计费信息，或 Together 成本分析/账户支持进行核对；
5. 将尝试归类为已成功、已终止失败、已取消或未解决；
6. 只要仍可能产生费用/输出且状态未解决，就必须在创建新任务前重新获得批准。

取消不等于从未运行。Replicate 文档说明，官方模型的取消操作仍可能产生费用；按时长计费的运行会收取已执行时间的费用。fal 提供取消 URL，但应查看所选模型的当前计费文档。Together 的同步图像路由未公开提供取消操作。

故障转移必须保留意图，而非参数

对于受治理的生产环境，禁用静默的 fal 故障转移。对于由应用管理的备用网关：

• 预先批准第二个模型、架构、发布方／许可证、数据路径、价格上限和验收基线；
• 只有在主请求已进入终态或财务结算完成后，才进行故障转移；
• 创建新的、针对特定提供方的负载，而不是仅重命名字段；
• 不要指望种子、安全决策、宽高比预设、图像数量、负面提示词或参考强度能够迁移；
• 标记产物实际使用的网关／模型／版本，不要在不作说明的情况下将其混入回归基线；
• 重新运行语义、几何、文本、相似度和策略检查。

如果需要精确的视觉可复现性，则无法进行故障转移。请改用固定的模型构建版本／部署，以及经过验证的确定性配置。

输入与产物保管

不要仅仅因为参考素材有公开 URL 就发送它。请核实版权／许可证、隐私／公开权同意、生物特征／儿童／受监管状态，以及允许的转换方式。保留源文件哈希和权利记录。

网关特定的输入注意事项：

• fal SDK 上传会生成公开的 CDN URL；负载删除 API 不会删除输入 CDN 文件。除非通过合同确认存在合适的私有企业级路径，否则请避免使用机密参考素材。
• Replicate 客户端接受托管 URL、不超过 100 MB 的本地文件，或仅建议用于小于 1 MB 内容的 data URI。API 预测数据通常会在一小时后删除，但仍须遵守市场平台／第三方条款。
• 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 保留期限由单独设置控制；数据会交由供应商／服务提供商处理；团队成员可以查看请求。若无适用合同／模型路径的规定，则客户输入是否用于训练、具体处理区域以及 Partner API 的保留期限均为未知。
• Replicate 说明 API 预测数据会在一小时后清理，且其分处理方位于美国。其条款授权服务处理数据并使用由此产生的数据；市场中的模型可能涉及额外的处理方／条款。模型页面上的“Zero training”并非适用于整个平台的保证。
• Together 的隐私政策称，训练用途须选择加入，并说明了 ZDR；其隐私／安全文档称默认不存储输入／输出。由于启用方式的表述存在冲突，应核实设置和合同，不要推断默认状态。可通过安排获得企业级数据驻留／私有网络；统一适用区域为未知。

对于个人、机密、受监管或受数据驻留要求约束的内容，必须完成 DPA 签署／安全审查，并确切确认路由方式。网关的一般性政策本身无法说明下游专有模型路径的情况。

发布与计费台账

只有在台账记录包含以下内容后，才能将运行标记为完成：

• 本地尝试 ID；网关请求／预测／响应 ID；时间戳和终止状态；
• 请求的网关／端点／模型／版本，以及返回的网关／模型／版本；
• 架构 URL／哈希／日期和客户端／SDK 版本；
• 负载哈希、种子、数量、请求／实际尺寸和格式；
• 价格来源／时间、计费单位、预估上限、可用时的实际用量记录、额度／折扣、审批 ID；
• 重试／回退／故障转移／取消事件及尚未解决的歧义；
• 保留期限／ZDR／store-IO 设置、执行／数据路径分类、权利／条款快照；
• 内容审核和人工审查处置情况；
• 持久化产物哈希、字节大小、解码元数据、来源核验结果及删除状态。

如果请求的身份与返回的身份不一致、架构／价格发生变化、出现无法对应已知请求 ID 的扣费、请求成功但没有产物、出现重复扣费／输出，或未经审批就生成产物，则发出告警。

一手来源列表

核查日期：2026-07-10：

fal.ai

• 模型 API 概览
• 异步队列
• 可靠性、重试和回退
• 平台标头和数据保留控制
• Webhook 验证
• 定价和定价 API
• FLUX Schnell 架构
• 常见问题：公开媒体和特定模型的许可证
• API 服务条款、条款和隐私政策

Replicate

• HTTP 预测 API 和状态
• 官方模型和社区模型
• 模型／版本架构发现
• 创建预测和截止时间
• 输入文件、输出文件和数据保留
• Webhook 投递和签名验证
• 计费、定价和速率限制
• FLUX Schnell 模型／API 页面、条款、隐私政策和子处理方

Together AI

• 图像端点参考
• 图像概览、参数和参考图像
• 当前无服务器模型目录和模型元数据 API
• 模型生命周期／弃用
• 动态速率限制、计费分析和定价
• 隐私与安全、隐私政策和服务条款
• FLUX Schnell 模型页面

来源之间存在冲突时，不要默默选择更方便的说法。应记录冲突，采用范围更窄、更安全的行为，并在进行实时或敏感工作前确认账户／合同条款。