xai-grok-imagine-video

xAI Grok Imagine の動画

直接 xAI API を、管理されたメディア制作サービスとして使用します。クリエイティブブリーフとソースの権利を確定し、価格を算出したリクエストを作成して、有料 POST の前に意図を永続化し、その後、非同期の結果を照合して、成果物とその証拠を保管します。

以下の変動する情報と価格は、すべて 2026-07-10 に xAI の一次情報で確認済みです。

適切な制作経路を選択する

直接利用するエンドポイントは https://api.x.ai です。サーバー側で Authorization: Bearer $XAI_API_KEY を使用して認証します。API キーは xAI Console のチームに属し、前払いクレジットまたは請求書払いの API 課金を利用します。

以下の別々のサービスを混同しないでください。

• grok.com とモバイルアプリの Grok は、コンシューマー向けプラン、週単位の利用枠、コンシューマー向けデータ管理、コンシューマー向け利用規約を使用します。「Video 1.5 Fast」はそこで宣伝されていますが、直接 API で使用するモデルのスラッグではありません。
• X 上の Grok には、X の利用規約とプライバシールールが適用されます。X Premium または SuperGrok のサブスクリプションは、API クレジットではありません。
• Grok Imagine の画像モデルは /v1/images/* を使用します。音声とスピーチには別のエンドポイントを使用します。それらのパラメーターと価格は、動画には適用されません。
• xAI のマーケティングで名前が挙げられているサービスを含むパートナープラットフォームやゲートウェイは、xAI モデルをラップしている場合がありますが、独自のスキーマ、モデレーション、データ保持、価格設定、リージョンを追加します。
• Azure、Google Cloud、その他のマネージドサービスでは、認証、クォータ、機能の同等性、課金、データガバナンスの契約がそれぞれ異なります。このスキルでは、xAI の直接 API について説明します。

公開 API では api.x.ai を使用します。xAI は複数の推論クラスターとエンタープライズ向けのデータレジデンシーオプションを提供しており、ステータスページには米国および EU のリージョン別 API サービスが表示されています。ページのラベルだけを根拠に、リージョンのホスト名を選択したり、データレジデンシーを保証したりしないでください。xAI Console または注文書／DPA でチームに提供されているエンドポイント、リージョン、契約上の管理手段のみを使用してください。

モデルとモードの選択ガイド

リクエストごとに有効にできるモードは1つだけです。

目的	エンドポイントとRESTフィールド	有効なモデル	出力の制御
テキストから動画を生成	POST /v1/videos/generations；prompt	grok-imagine-video	duration は1～15秒；指定されたアスペクト比；480p／720p
画像から動画を生成	generations；prompt + image	どちらのモデルでも可	1～15秒；指定されたアスペクト比；基本モデル：480p／720p；1.5：480p／720p／1080p
参照画像から動画を生成	generations；prompt + reference_images	grok-imagine-video のみ	1～10秒；指定されたアスペクト比；480p／720p；参照画像は1～7枚
動画を編集	POST /v1/videos/edits；prompt + video	grok-imagine-video のみ	カスタムの長さ／アスペクト比／解像度は指定不可；入力は8.7秒以下；出力は長さとアスペクト比を引き継ぎ、解像度は最大720p
動画を延長	POST /v1/videos/extensions；prompt + video	grok-imagine-video のみ	入力は2～15秒；延長は2～10秒、デフォルトは6秒；出力はアスペクト比と解像度を引き継ぎ、解像度は最大720p

grok-imagine-video-1.5 は、一般提供（GA）されている画像から動画を生成する機能専用です。テキストから動画を生成する機能や参照画像から動画を生成する機能には対応しておらず、モデルページで宣言されている入力は画像のみです。現在のエイリアスは grok-imagine-video-1.5-preview と grok-imagine-video-1.5-2026-05-30 です。記録済みの再現性に関する判断で日付付きエイリアスが必要な場合を除き、GAのスラッグを使用してください。コンシューマー向けの「1.5 Fast」という表記をAPI識別子と混同しないでください。

生成で許可されているアスペクト比は 1:1、16:9、9:16、4:3、3:4、3:2、2:3 で、デフォルトは 16:9 です。画像から動画を生成する場合、アスペクト比を省略すると入力画像のアスペクト比が使われます。aspect_ratio を指定するとこれが上書きされ、画像は引き伸ばされます。歪みが許容できない場合は、送信前に意図的にトリミングまたはパディングしてください。デフォルトの解像度は 480p です。1080pを選択できるのは、1.5で画像から動画を生成する場合に限られます。

参照画像は、人、製品、衣服、スタイルなどを指定するために使われ、最初のフレームにはなりません。RESTの形式は次のとおりです。

{
  "model": "grok-imagine-video",
  "prompt": "The cyclist from <IMAGE_1> rides past the cafe from <IMAGE_2>.",
  "reference_images": [
    {"file_id": "file_authorized_subject"},
    {"url": "https://authorized.example/cafe.jpg"}
  ],
  "duration": 8,
  "aspect_ratio": "16:9",
  "resolution": "720p"
}

image、reference_images、video のいずれかを使用し、混在させないでください。image + reference_images は明示的に無効です。参照画像にはPNG／JPEG／WebP、動画にはMP4を使用できます。Files経由で参照する場合、各ロケーターにはHTTPS URL、Base64データURI、またはプライベートな file_id を指定できます。現在の一般的なFilesガイドには48 MB、アップロードリファレンスには50 MBと記載されています。許容される最大サイズはドキュメント間で矛盾しているため、48 MB未満に抑えるか、アップロード前に正確なエンドポイントを確認してください。Files以外での公開メディア／Base64の上限は記載されていません。

HTTPSロケーターを指定すると、xAIがそのリソースを取得できるようになり、リクエストの処理やログ記録にURLが露出します。クエリ文字列にベアラー認証情報のような秘密情報が含まれる場合があります。繰り返し使用する場合は、意図したTTLを設定したプライベートな file_id を優先してください。Base64データや署名付きURLをログに記録しないでください。また、DNS／IP／リダイレクトに関するSSRF対策とバイト数／時間の制限なしに、独自アプリケーションで任意のユーザーURLをプロキシしたり取得したりしないでください。

xAIのローンチ発表と1.5の発表では、アクションに同期した効果音、環境音、台詞などのネイティブな音声生成が説明されています。これはプロバイダーの主張です。現在の動画RESTスキーマには、音声の有効／無効を切り替えるフィールド、個別の音声プロンプト、言語設定、サンプルレート、チャンネル数、コーデック、ラウドネスが記載されていません。フレームレートについても同様に記載がありません。必要な音や台詞はメインのプロンプトに控えめに含め、実際のストリームを確認してください。結果が得られる前に、音声の有無や品質を保証しないでください。

メディア変換全体の料金を見積もる

現在の直接 API の USD 料金：

モデル	メディア入力	480p 出力	720p 出力	1080p 出力
grok-imagine-video	画像 $0.002/枚、動画 $0.01/ソース秒	$0.05/出力秒	$0.07/出力秒	非対応
grok-imagine-video-1.5	画像 $0.01	$0.08/出力秒	$0.14/出力秒	$0.25/出力秒

テキスト入力には、記載されているメディア料金はありません。承認前に計算してください。

• T2V：duration × output rate。
• I2V：画像料金 1 枚分に duration × output rate を加算します。
• 参照：number of reference images × $0.002 に duration × output rate を加算します。
• 編集：source duration × $0.01 の動画入力料金に、output duration × inherited/capped-resolution output rate を加算します。出力時間はソース時間と同じです。
• 延長：source duration × $0.01 に、extension duration × inherited/capped-resolution output rate を加算します。xAI は延長の料金を期間限定のプロモーション料金であり、変更される可能性があるとしています。承認の直前に必ず再確認してください。

例：ベースモデルで 6 秒の 720p T2V は $0.42、1.5 で 12 秒の 1080p I2V は $0.01 + $3.00 = $3.01、ベースモデルで 8 秒の 720p 参照リクエスト（画像 3 枚）は $0.006 + $0.56 = $0.566 です。各バリアントは、個別に課金される可能性のあるリクエストとして見積もってください。メディアのバッチリクエストにバッチ割引はありません。

完了後、usage.cost_in_usd_ticks がリクエストごとの実際の料金です。10,000,000,000 ティックは USD 1 に相当します。整数値を保存し、承認済みの見積もりと比較してください。料金の見積もりは、承認なしの超過請求を許可するものではありません。

名詞を飾るのではなく、動きを直接指定する

xAI は自然言語の例を公開していますが、正式なカメラコマンド文法は公開していません。角括弧のコマンド、ネガティブプロンプトのフィールド、シード、CFG、FPS、タイムコードの保証を勝手に作らないでください。

生成プロンプトは、次の 5 つの要素を中心に組み立てます。

1. 冒頭の構図と、維持する必要のあるアイデンティティや素材の特性を固定します。
2. 重さや接触が目に見える、因果関係のある動作の流れを被写体に 1 つ与えます。
3. 通常の撮影用語で、主となるカメラの動きを 1 つ指定します。例：ゆっくり寄る、右にパンする、ドリーアウトする、上にチルトする、固定する、手持ちで追う。
4. 「最初に」「次に」と観察可能な終わりを使い、少数のビートだけを順序立てます。
5. 照明やスタイルを加え、重要であれば短い音やセリフの指示も加えます。

これは制作上のヒューリスティックであり、パラメーターの仕様ではありません。8 秒の簡潔な指示例：「雨で濡れたカフェのテーブルに置かれた青い陶器のカップを、夜明けの広角ショットで映す。手が画面に入り、カップを持ち上げる。湯気がレンズに向かって渦を巻き、カメラはゆっくり一度だけ寄る。窓辺でカップを静止させたところで終わる。静かな室内音と、陶器が軽く触れ合う音を 1 回。」形容詞を鵜呑みにせず、動作の因果関係、オクルージョン、手、文字、反射、音声同期を確認してください。

モードごとの指示：

• 1.5 またはベースモデルの I2V： 画像が開始フレームになります。何が動き始めるか、カメラがどのように移動するか、どの状態で終わるかなど、変化を説明してください。画像と矛盾する構図を改めて説明するのは避けてください。互換性のないアスペクト比に引き伸ばすのではなく、API の外部でクロップまたはパディングしてください。
• 参照画像からの動画生成： xAI の例のように、順序付きの各参照画像に 1 つずつ役割を割り当て、プロンプト内で <IMAGE_1>、<IMAGE_2> などを使ってください。どのアイデンティティがどの製品を身に着けるか、またどの参照画像が環境やスタイルを指定するかを明示してください。関連性が曖昧な 7 枚より、互いに矛盾しない少数の参照画像のほうが監査しやすくなります。
• 編集： 変更を最小限にするよう求め、不変条件を明示してください。「傘だけをマットな黄色に変更し、顔、タイミング、カメラの動き、雨、看板、音声は維持してください。」xAI は高い保持性能をうたっていますが、編集全体を元の素材と比較するまでは、あくまでプロバイダーの主張です。
• 延長： 最後に見える動き、カメラの速度、照明、音のベッドから始めます。次の動作だけを説明してください。返されるファイルには元の動画と延長部分の両方が含まれます。つなぎ目、重複フレーム、リズム、アイデンティティ、音声の切り替わりを確認してください。

リクエスト送信前に有料利用の境界を確立する

エンドポイントと本文の正確な内容、モデルまたは日付付きエイリアス、価格ページのタイムスタンプ、見積もり式、ソースの長さ／解像度、バリアント数、プロンプト、アスペクト比／解像度、ソースのハッシュと権利、対象のリージョン／チーム、保持／公開共有の選択、モデレーション／開示のレビュー担当者を含む記録を作成します。

キーが不要で、ネットワークリクエストを行わず、排他的な作成台帳を消費しないローカルのドライランをデフォルトにします。プロンプト、ファイル ID、Base64、完全な署名付き URL はハッシュに置き換えて秘匿します。正確なエンドポイント／本文のダイジェスト、モデル／モード、順序付きソースハッシュ、ローカルジョブキー、入力／出力の正確なホストポリシー、保存先、最新の価格根拠、ZDR 要件、権利／モデレーション／QA の根拠、USD の見積額、作成回数 1 回を正規形式で関連付けます。対応するダイジェストと、有限かつ正の USD 上限額を必須とします。POST する前に、排他的な作成セマンティクスを用いてインテントレコードを書き込みます。

動画エンドポイントでは、冪等性キーがドキュメントに記載されていません。そのため、作成 POST を自動で再試行してはいけません。タイムアウト、接続リセット、または 5xx が発生し、request_id を取得できなかった場合は、create-outcome-unknown と記録し、代替の作成を承認する前に xAI Console の請求情報、コスト記録、サポートを通じて照合します。既知の 4xx 拒否は修正して、新しい試行として再送信できます。認証済みのステータス GET と最終的なメディアのダウンロードでは、回数を制限したバックオフを使用できます。

service_unavailable は後で再試行可能と記載されていますが、これは有料の作成リクエストを冪等にするものではありません。その他の終端コードには、invalid_argument、permission_denied、failed_precondition があります。一般的な HTTP エラーには、400、401、403、404、415、422、429 があります。モデレーションを回避するために、ブロックされたプロンプトを変更してはいけません。

検証時に、ファーストパーティのレート制限に関するレンダリング結果は一致しませんでした。現在の制限／モデルの情報では、レンダリング／クラスターに応じて、動画のレート制限として 1 RPS、10 RPS、70 RPM が返されました。したがって、正確な運用上の制限は公開ドキュメントからは不明です。チーム固有の xAI Console の制限を正として、作成リクエストは保守的なレートで送信してください。この不一致を理由に、表示された最大値で並行処理を行ってはいけません。

例：実行可能なガード付き REST クライアント

この Python 3.11 以降の例を xai_video_guard.py として、スキルディレクトリの外に保存します。5 つすべての直接モードに対応し、メディア入力と出力の料金を両方見積もり、デフォルトではドライランを実行し、厳密なダイジェストと上限額を必須とします。また、作成前にインテントを保存し、作成を再試行せず、既知のリクエスト ID から処理を再開し、ZDR ヘッダーを記録し、モデレーションを検証して、xAI の実際のコストティックを取得します。ベアラーやリダイレクトを使わず、ホストを厳密に照合し、公開 DNS に固定した TLS 接続を通じてダウンロードし、MP4 を永続的にステージングし、プローブして完全にデコードしたうえで、クラッシュ後も再開可能な方法で公開します。Python 3.11 以降、ffprobe、ffmpeg が必要です。

すべてのメディアロケーターについて、--source-local で順序付きのローカルミラーを指定します。プログラムは各ミラーをハッシュ化して完全にデコードします。編集／延長の場合は、長さと解像度を再度プローブし、ドシエの不一致があれば拒否します。ソースが 1 つの場合はファイルの SHA-256 を使用します。参照が複数ある場合は、ドシエ処理で出力された正規化済み順序付きリストのダイジェストを使用します。HTTPS ロケーターには、厳密な --input-host の指定も必要です。非公開の file_... ID には必要ありません。

#!/usr/bin/env python3
import argparse
import hashlib
import http.client
import ipaddress
import json
import os
import pathlib
import random
import socket
import ssl
import subprocess
import sys
import tempfile
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation

API = "https://api.x.ai/v1"
OUT_RATE = {
    "grok-imagine-video": {"480p": Decimal("0.05"), "720p": Decimal("0.07")},
    "grok-imagine-video-1.5": {"480p": Decimal("0.08"), "720p": Decimal("0.14"), "1080p": Decimal("0.25")},
}
IMAGE_RATE = {"grok-imagine-video": Decimal("0.002"), "grok-imagine-video-1.5": Decimal("0.01")}
VIDEO_INPUT_RATE = Decimal("0.01")
ASPECTS = {"1:1", "16:9", "9:16", "4:3", "3:4", "3:2", "2:3"}
MAX_JSON = 256 * 1024
MAX_VIDEO = 512 * 1024 * 1024
MAX_TOOL_OUTPUT = 512 * 1024

class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl): return None

NO_REDIRECT = urllib.request.build_opener(NoRedirect)

class APIProblem(Exception):
    def __init__(self, kind, status=None, body_sha256=None):
        super().__init__(kind); self.kind, self.status, self.body_sha256 = kind, status, body_sha256

def canonical(value): return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
def sha256_bytes(value): return hashlib.sha256(value).hexdigest()
def utc_now(): return datetime.now(timezone.utc).isoformat()

def fsync_directory(path):
    if os.name != "nt":
        fd = os.open(path, os.O_RDONLY)
        try: os.fsync(fd)
        finally: os.close(fd)

def save_record(path, record):
    target = pathlib.Path(path); target.parent.mkdir(parents=True, exist_ok=True)
    fd, name = tempfile.mkstemp(prefix=".xai-video-", suffix=".json", dir=target.parent)
    try:
        os.fchmod(fd, 0o600)
        with os.fdopen(fd, "wb") as out:
            out.write(canonical(record) + b"\n"); out.flush(); os.fsync(out.fileno())
        os.replace(name, target); fsync_directory(target.parent)
    finally:
        if os.path.exists(name): os.unlink(name)

def exclusive_record(path, record):
    target = pathlib.Path(path); target.parent.mkdir(parents=True, exist_ok=True)
    fd = os.open(target, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
    with os.fdopen(fd, "wb") as out:
        out.write(canonical(record) + b"\n"); out.flush(); os.fsync(out.fileno())
    fsync_directory(target.parent)

def read_record(path):
    raw = pathlib.Path(path).read_bytes()
    if len(raw) > MAX_JSON: raise SystemExit("Ledger exceeds JSON cap")
    value = json.loads(raw)
    if not isinstance(value, dict): raise SystemExit("Ledger is not an object")
    return value

def bounded_process(argv, timeout, cap=MAX_TOOL_OUTPUT):
    process = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    chunks, total, overflow, lock = [], [0], threading.Event(), threading.Lock()
    def drain(stream):
        while True:
            data = stream.read(65536)
            if not data: return
            with lock:
                total[0] += len(data)
                if total[0] > cap: overflow.set(); process.kill(); return
                chunks.append(data)
    threads = [threading.Thread(target=drain, args=(stream,), daemon=True) for stream in (process.stdout, process.stderr)]
    for thread in threads: thread.start()
    try: code = process.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        process.kill(); process.wait(); raise RuntimeError(f"tool timeout: {pathlib.Path(argv[0]).name}")
    for thread in threads: thread.join(timeout=2)
    output = b"".join(chunks)
    if overflow.is_set(): raise RuntimeError(f"tool output cap exceeded: {pathlib.Path(argv[0]).name}")
    if code: raise RuntimeError(f"tool failed: {pathlib.Path(argv[0]).name}; diagnostics_sha256={sha256_bytes(output)}")
    return output

def exact_hosts(raw, label):
    hosts = []
    for item in raw.split(","):
        host = item.strip().rstrip(".").lower()
        if not host: continue
        if "*" in host or "/" in host or ":" in host: raise SystemExit(f"{label} accepts exact DNS hostnames only")
        try: ipaddress.ip_address(host)
        except ValueError: pass
        else: raise SystemExit(f"{label} rejects IP literals")
        hosts.append(host)
    if not hosts: raise SystemExit(f"{label} requires at least one reviewed exact host")
    return tuple(sorted(set(hosts)))

def public_addresses(host):
    addresses = []
    for item in socket.getaddrinfo(host, 443, type=socket.SOCK_STREAM):
        address = ipaddress.ip_address(item[4][0])
        if not address.is_global: raise SystemExit("Media host resolves to a non-public address")
        addresses.append(str(address))
    if not addresses: raise SystemExit("Media host did not resolve")
    return sorted(set(addresses))

def locator(value, allowed_hosts):
    if value.startswith("file_"):
        if len(value) > 256 or any(character not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-" for character in value):
            raise SystemExit("Invalid xAI file_id")
        return {"file_id": value}
    parsed = urllib.parse.urlsplit(value)
    if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password or parsed.fragment or parsed.port not in (None, 443):
        raise SystemExit("Media locator must be an xAI file_id or public HTTPS URL")
    host = parsed.hostname.rstrip(".").lower()
    try:
        ipaddress.ip_address(host)
    except ValueError: pass
    else: raise SystemExit("IP-literal media URLs are forbidden")
    if host not in allowed_hosts: raise SystemExit("Media URL host is outside the approved exact-host policy")
    return {"url": value}

def redact(value):
    if "url" in value:
        p = urllib.parse.urlsplit(value["url"])
        return {"url_host": p.hostname, "url_sha256": sha256_bytes(value["url"].encode("utf-8"))}
    if "file_id" in value: return {"file_id_sha256": sha256_bytes(value["file_id"].encode("utf-8"))}
    return {"locator_sha256": sha256_bytes(canonical(value))}

def request_json(method, url, api_key, payload=None, retries=0):
    body = None if payload is None else canonical(payload)
    for attempt in range(retries + 1):
        req = urllib.request.Request(url, data=body, method=method, headers={
            "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", "Accept": "application/json"})
        try:
            with NO_REDIRECT.open(req, timeout=45) as response:
                raw = response.read(MAX_JSON + 1)
                zdr = response.headers.get("x-zero-data-retention")
            if len(raw) > MAX_JSON: raise APIProblem("response-too-large")
            try: data = json.loads(raw)
            except json.JSONDecodeError: raise APIProblem("invalid-json", response.status, sha256_bytes(raw))
            if not isinstance(data, dict): raise APIProblem("non-object-json", response.status, sha256_bytes(raw))
            return data, None if zdr is None else zdr.lower()
        except urllib.error.HTTPError as exc:
            raw = exc.read(MAX_JSON + 1)
            problem = APIProblem("http-error", exc.code, sha256_bytes(raw))
            if 400 <= exc.code < 500 or attempt == retries:
                raise problem
        except APIProblem: raise
        except (urllib.error.URLError, TimeoutError, OSError) as exc:
            if attempt == retries: raise APIProblem("transport-" + type(exc).__name__)
        time.sleep(min(2 ** attempt, 8) * (1 + random.random() * 0.25))

def source_evidence(args):
    required = 0 if args.mode == "text" else (len(args.reference) if args.mode == "reference" else 1)
    if len(args.source_local) != required: raise SystemExit(f"Provide exactly {required} ordered --source-local mirrors")
    evidence = []
    for path_text in args.source_local:
        path = pathlib.Path(path_text).resolve()
        if not path.is_file() or path.stat().st_size <= 0 or path.stat().st_size >= 48 * 1024 * 1024:
            raise SystemExit("Source mirror must exist and remain below the conservative 48 MiB Files limit")
        digest = hashlib.sha256()
        with path.open("rb") as stream:
            header = stream.read(32); stream.seek(0)
            for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk)
        kind = "video" if args.mode in {"edit", "extend"} else "image"
        if kind == "image" and not (header.startswith(b"\xff\xd8\xff") or header.startswith(b"\x89PNG\r\n\x1a\n") or (header.startswith(b"RIFF") and header[8:12] == b"WEBP")):
            raise SystemExit("Image mirror must be JPEG, PNG, or WebP by signature")
        if kind == "video" and (len(header) < 12 or header[4:8] != b"ftyp"):
            raise SystemExit("Video mirror must be MP4 by signature")
        raw = bounded_process(["ffprobe", "-v", "error", "-show_streams", "-show_format", "-of", "json", str(path)], 30, MAX_JSON)
        metadata = json.loads(raw); streams = metadata.get("streams", [])
        visual = next((item for item in streams if item.get("codec_type") == "video"), None)
        if not visual: raise SystemExit("Source mirror has no visual stream")
        bounded_process(["ffmpeg", "-nostdin", "-v", "error", "-i", str(path), "-f", "null", "-"], 300)
        item = {"path": str(path), "sha256": digest.hexdigest(), "bytes": path.stat().st_size,
                "kind": kind, "width": visual.get("width"), "height": visual.get("height")}
        if kind == "video":
            try: duration = Decimal(str(metadata.get("format", {}).get("duration")))
            except InvalidOperation: raise SystemExit("Source video duration is invalid")
            if not duration.is_finite() or duration <= 0: raise SystemExit("Source duration must be finite and positive")
            short_edge = min(int(visual.get("width")), int(visual.get("height")))
            resolution = "480p" if short_edge <= 480 else ("720p" if short_edge <= 720 else "1080p")
            item.update(duration=str(duration), codec=visual.get("codec_name"), resolution=resolution)
        evidence.append(item)
    if evidence and args.source_sha256:
        expected = evidence[0]["sha256"] if len(evidence) == 1 else sha256_bytes(canonical([item["sha256"] for item in evidence]))
        if args.source_sha256.lower() != expected: raise SystemExit("source-sha256 does not match the validated local mirror evidence")
    if evidence and args.mode in {"edit", "extend"}:
        actual_duration = Decimal(evidence[0]["duration"])
        if args.source_duration is not None and abs(args.source_duration - actual_duration) > Decimal("0.05"):
            raise SystemExit("Declared source duration differs from ffprobe evidence")
        if args.source_resolution is not None and args.source_resolution != evidence[0]["resolution"]:
            raise SystemExit("Declared source resolution differs from ffprobe evidence")
        args.source_duration, args.source_resolution = actual_duration, evidence[0]["resolution"]
    return evidence

def quote(args):
    if args.mode == "edit":
        output_seconds = Decimal(str(args.source_duration))
        output_resolution = "720p" if args.source_resolution in {"720p", "1080p"} else "480p"
        return output_seconds * (VIDEO_INPUT_RATE + OUT_RATE[args.model][output_resolution])
    if args.mode == "extend":
        output_resolution = "720p" if args.source_resolution in {"720p", "1080p"} else "480p"
        return Decimal(str(args.source_duration)) * VIDEO_INPUT_RATE + Decimal(args.duration) * OUT_RATE[args.model][output_resolution]
    result = Decimal(args.duration) * OUT_RATE[args.model][args.resolution]
    if args.mode == "image":
        result += IMAGE_RATE[args.model]
    elif args.mode == "reference":
        result += IMAGE_RATE[args.model] * len(args.reference)
    return result

def validate(args):
    if not args.prompt.strip(): raise SystemExit("A non-empty prompt is required")
    if args.model == "grok-imagine-video-1.5" and args.mode != "image":
        raise SystemExit("grok-imagine-video-1.5 is image-to-video only")
    if args.mode in {"text", "image", "reference"}:
        if args.duration is None or not 1 <= args.duration <= 15: raise SystemExit("Generation duration must be 1–15 seconds")
        if args.mode == "reference" and args.duration > 10: raise SystemExit("Reference-to-video maximum is 10 seconds")
        if args.resolution not in OUT_RATE[args.model]: raise SystemExit("Resolution unsupported by selected model")
        if args.aspect_ratio not in ASPECTS: raise SystemExit("Unsupported aspect ratio")
    if args.mode == "image" and not args.image: raise SystemExit("--image is required")
    if args.mode == "reference" and not 1 <= len(args.reference) <= 7: raise SystemExit("Provide 1–7 references")
    if args.mode in {"edit", "extend"}:
        if not args.video or args.source_duration is None or args.source_resolution is None:
            raise SystemExit("Video mode requires --video, --source-duration, --source-resolution, and a local mirror")
        if args.model != "grok-imagine-video": raise SystemExit("Editing and extension require grok-imagine-video")
        if not args.source_duration.is_finite(): raise SystemExit("Source duration must be finite")
    if args.mode == "edit" and not 0 < args.source_duration <= 8.7:
        raise SystemExit("Edit input must be at most 8.7 seconds")
    if args.mode == "extend":
        if not 2 <= args.source_duration <= 15: raise SystemExit("Extension input must be 2–15 seconds")
        if args.duration is None or not 2 <= args.duration <= 10: raise SystemExit("Extension duration must be 2–10 seconds")
    if args.mode != "text" and not args.source_sha256:
        raise SystemExit("Media mode requires --source-sha256 provenance evidence")
    if args.source_sha256 and (len(args.source_sha256) != 64 or any(c not in "0123456789abcdefABCDEF" for c in args.source_sha256)):
        raise SystemExit("--source-sha256 must be exactly 64 hexadecimal characters")

def build(args, input_hosts):
    endpoint = "/videos/generations"
    body = {"model": args.model, "prompt": args.prompt}
    if args.mode in {"text", "image", "reference"}:
        body.update(duration=args.duration, aspect_ratio=args.aspect_ratio, resolution=args.resolution)
    if args.mode == "image": body["image"] = locator(args.image, input_hosts)
    if args.mode == "reference": body["reference_images"] = [locator(x, input_hosts) for x in args.reference]
    if args.mode == "edit": endpoint = "/videos/edits"; body["video"] = locator(args.video, input_hosts)
    if args.mode == "extend":
        endpoint = "/videos/extensions"; body["video"] = locator(args.video, input_hosts); body["duration"] = args.duration
        body.pop("aspect_ratio", None); body.pop("resolution", None)
    return endpoint, body

def redacted_body(body):
    clean = dict(body)
    for key in ("image", "video"):
        if key in clean: clean[key] = redact(clean[key])
    if "reference_images" in clean: clean["reference_images"] = [redact(x) for x in clean["reference_images"]]
    return clean

class PinnedHTTPSConnection(http.client.HTTPSConnection):
    def __init__(self, host, address, timeout):
        super().__init__(host, 443, timeout=timeout, context=ssl.create_default_context()); self.address = address
    def connect(self):
        raw = socket.create_connection((self.address, 443), self.timeout)
        try: self.sock = self._context.wrap_socket(raw, server_hostname=self.host)
        except Exception: raw.close(); raise

def output_target(url, allowed_hosts):
    parsed = urllib.parse.urlsplit(url)
    if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password or parsed.fragment or parsed.port not in (None, 443):
        raise RuntimeError("Output URL violates HTTPS authority policy")
    host = parsed.hostname.rstrip(".").lower()
    try: ipaddress.ip_address(host)
    except ValueError: pass
    else: raise RuntimeError("Output URL rejects IP literals")
    if host not in allowed_hosts: raise RuntimeError("Output host is outside the approved exact-host policy")
    addresses = public_addresses(host); target = parsed.path or "/"
    if parsed.query: target += "?" + parsed.query
    return host, target, addresses

def verify_mp4(path):
    path = pathlib.Path(path); size = path.stat().st_size
    if size <= 0 or size > MAX_VIDEO: raise RuntimeError("Artifact size is outside policy")
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        if stream.read(12)[4:8] != b"ftyp": raise RuntimeError("Artifact lacks MP4 ftyp signature")
        stream.seek(0)
        for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk)
    raw = bounded_process(["ffprobe", "-v", "error", "-show_streams", "-show_format", "-of", "json", str(path)], 30, MAX_JSON)
    metadata = json.loads(raw); streams = metadata.get("streams", [])
    video = next((item for item in streams if item.get("codec_type") == "video"), None)
    if not video: raise RuntimeError("Artifact has no video stream")
    bounded_process(["ffmpeg", "-nostdin", "-v", "error", "-i", str(path), "-f", "null", "-"], 300)
    return {"bytes": size, "sha256": digest.hexdigest(), "qa": {"container": metadata.get("format", {}).get("format_name"),
            "video_codec": video.get("codec_name"), "width": video.get("width"), "height": video.get("height"),
            "fps": video.get("avg_frame_rate"), "duration": metadata.get("format", {}).get("duration"),
            "audio_present": any(item.get("codec_type") == "audio" for item in streams), "full_decode": "passed"}}

def download_mp4(url, output, allowed_hosts, ledger_path, record):
    target = pathlib.Path(output); target.parent.mkdir(parents=True, exist_ok=True)
    stage = target.with_name("." + target.name + ".xai-stage"); expected = record.get("artifact_staged") or record.get("artifact")
    if target.exists():
        if not isinstance(expected, dict): raise RuntimeError("Existing artifact has no durable expected evidence")
        artifact = verify_mp4(target)
        if (artifact["sha256"], artifact["bytes"]) != (expected.get("sha256"), expected.get("bytes")): raise RuntimeError("Existing artifact mismatch")
        return artifact
    if stage.exists() and isinstance(expected, dict):
        artifact = verify_mp4(stage)
        if (artifact["sha256"], artifact["bytes"]) != (expected.get("sha256"), expected.get("bytes")): raise RuntimeError("Staged artifact mismatch")
        os.replace(stage, target); fsync_directory(target.parent); return artifact
    host, path_query, addresses = output_target(url, allowed_hosts); url_hash = sha256_bytes(url.encode("utf-8"))
    if record.get("output_url_sha256") not in (None, url_hash): raise RuntimeError("Output URL changed during incomplete transfer")
    record.update(state="downloading", output_url_sha256=url_hash, output_host=host, download_started_at=record.get("download_started_at") or utc_now())
    save_record(ledger_path, record); stage.unlink(missing_ok=True)
    connection = PinnedHTTPSConnection(host, addresses[0], 120)
    try:
        connection.request("GET", path_query, headers={"Host": host, "Accept": "video/mp4", "User-Agent": "xai-video-guard/2"})
        response = connection.getresponse()
        if response.status != 200: raise RuntimeError(f"Output returned HTTP {response.status}")
        media_type = response.getheader("Content-Type", "").split(";", 1)[0].lower()
        if media_type not in {"video/mp4", "application/octet-stream"}: raise RuntimeError("Unexpected output media type")
        announced = response.getheader("Content-Length")
        if announced is not None and (not announced.isdigit() or int(announced) > MAX_VIDEO): raise RuntimeError("Invalid/excessive Content-Length")
        fd = os.open(stage, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600); total = 0
        try:
            with os.fdopen(fd, "wb") as dst:
                while True:
                    chunk = response.read(1024 * 1024)
                    if not chunk: break
                    total += len(chunk)
                    if total > MAX_VIDEO: raise RuntimeError("Output exceeded byte cap")
                    dst.write(chunk)
                dst.flush(); os.fsync(dst.fileno())
        except BaseException: stage.unlink(missing_ok=True); raise
    finally: connection.close()
    artifact = verify_mp4(stage); record.update(state="artifact-staged", artifact_staged=artifact, artifact_validated_at=utc_now())
    save_record(ledger_path, record); os.replace(stage, target); fsync_directory(target.parent); return artifact

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--mode", choices=["text", "image", "reference", "edit", "extend"], required=True)
    parser.add_argument("--model", choices=list(OUT_RATE), required=True)
    parser.add_argument("--prompt", required=True); parser.add_argument("--duration", type=int)
    parser.add_argument("--resolution", choices=["480p", "720p", "1080p"], default="480p")
    parser.add_argument("--aspect-ratio", default="16:9")
    parser.add_argument("--image"); parser.add_argument("--reference", action="append", default=[]); parser.add_argument("--video")
    parser.add_argument("--source-duration", type=Decimal); parser.add_argument("--source-resolution", choices=["480p", "720p", "1080p"])
    parser.add_argument("--source-sha256", help="SHA-256 of source file, or a dossier hash covering ordered references")
    parser.add_argument("--source-local", action="append", default=[], help="Ordered local mirror; repeat for references")
    parser.add_argument("--job-key", required=True)
    parser.add_argument("--input-host", default="", help="Comma-separated reviewed exact source URL hosts; omit for file_id-only inputs")
    parser.add_argument("--output-host", required=True, help="Comma-separated reviewed exact result hosts")
    parser.add_argument("--pricing-checked-at", required=True); parser.add_argument("--pricing-evidence-sha256", required=True)
    parser.add_argument("--rights-sha256", required=True); parser.add_argument("--moderation-sha256", required=True); parser.add_argument("--qa-plan-sha256", required=True)
    parser.add_argument("--ledger", required=True); parser.add_argument("--output", default="xai-video.mp4")
    parser.add_argument("--execute", action="store_true"); parser.add_argument("--approval-sha256"); parser.add_argument("--max-cost-usd")
    parser.add_argument("--require-zdr", action="store_true")
    args = parser.parse_args()
    if not args.job_key or any(not (character.isalnum() or character in "-_") for character in args.job_key): raise SystemExit("Invalid job-key")
    validate(args)
    input_hosts = () if not args.input_host.strip() else exact_hosts(args.input_host, "input-host")
    output_hosts = exact_hosts(args.output_host, "output-host")
    endpoint, body = build(args, set(input_hosts)); sources = source_evidence(args); amount = quote(args)
    if not amount.is_finite() or amount <= 0: raise SystemExit("Computed cost must be finite and positive")
    for label, value in (("pricing", args.pricing_evidence_sha256), ("rights", args.rights_sha256), ("moderation", args.moderation_sha256), ("QA", args.qa_plan_sha256)):
        if len(value) != 64 or any(character not in "0123456789abcdefABCDEF" for character in value): raise SystemExit(f"{label} evidence must be SHA-256")
    safe_body = redacted_body(body); safe_body["prompt"] = "sha256:" + sha256_bytes(args.prompt.encode("utf-8"))
    safe_sources = [{key: value for key, value in item.items() if key != "path"} | {"path_sha256": sha256_bytes(item["path"].encode("utf-8"))} for item in sources]
    output_policy = {"exact_hosts": list(output_hosts), "https": True, "redirects": "reject", "public_dns": True,
                     "reject_ip_literals": True, "dns_pinning": "connect-IP-with-TLS-SNI-and-host-certificate",
                     "max_bytes": MAX_VIDEO, "container": "mp4", "full_decode": True}
    request_hash = sha256_bytes(canonical(body))
    envelope = {"schema_version": 1, "job_key": args.job_key, "api_origin": API, "endpoint": endpoint,
                "mode": args.mode, "model": args.model, "request_sha256": request_hash, "redacted_request": safe_body,
                "sources": safe_sources, "input_exact_hosts": list(input_hosts), "output": str(pathlib.Path(args.output).resolve()),
                "output_policy": output_policy, "quoted_usd": format(amount, "f"), "create_count": 1,
                "pricing": {"checked_at": args.pricing_checked_at, "evidence_sha256": args.pricing_evidence_sha256.lower()},
                "governance": {"rights_sha256": args.rights_sha256.lower(), "moderation_sha256": args.moderation_sha256.lower(),
                               "qa_plan_sha256": args.qa_plan_sha256.lower(), "require_zdr": args.require_zdr}}
    approval = sha256_bytes(canonical(envelope))
    print(json.dumps({"approval_sha256": approval, "plan": envelope}, indent=2, ensure_ascii=False))
    print(f"DRY RUN — exact maximum USD {format(amount, 'f')}; one create; no network request sent", file=sys.stderr)
    if not args.execute: return
    if args.approval_sha256 != approval: raise SystemExit("Exact approval digest mismatch")
    try: maximum = Decimal(args.max_cost_usd or "")
    except InvalidOperation: raise SystemExit("max-cost-usd must be decimal")
    if not maximum.is_finite() or maximum <= 0 or maximum < amount: raise SystemExit("max-cost-usd must be finite, positive, and at least the quote")
    try: checked_at = datetime.fromisoformat(args.pricing_checked_at.replace("Z", "+00:00"))
    except ValueError: raise SystemExit("pricing-checked-at must be ISO-8601")
    if checked_at.tzinfo is None: raise SystemExit("pricing-checked-at must include timezone")
    age = datetime.now(timezone.utc) - checked_at.astimezone(timezone.utc)
    if age.total_seconds() < -300 or age.total_seconds() > 86400: raise SystemExit("Pricing check must be no more than 24 hours old")
    for item in body.values():
        locators = item if isinstance(item, list) else [item]
        for locator_item in locators:
            if isinstance(locator_item, dict) and "url" in locator_item: public_addresses(urllib.parse.urlsplit(locator_item["url"]).hostname)
    if args.require_zdr and os.environ.get("XAI_ZDR_CONFIRMED") != "1":
        raise SystemExit("Verify ZDR in xAI Console, then set XAI_ZDR_CONFIRMED=1 for this governed run")
    api_key = os.environ.get("XAI_API_KEY")
    if not api_key: raise SystemExit("Set XAI_API_KEY in the server-side environment")
    ledger_path = pathlib.Path(args.ledger)
    if ledger_path.exists():
        record = read_record(ledger_path)
        if record.get("request_sha256") != request_hash or record.get("approval_sha256") != approval: raise SystemExit("Ledger belongs to a different approved request")
        request_id = record.get("request_id")
        if not request_id: raise SystemExit(f"Ledger state {record.get('state')!r} has no request ID; never replay")
        zdr = record.get("create_zdr_header")
    else:
        record = {"schema_version": 1, "state": "intent-recorded", "recorded_at": utc_now(), "job_key": args.job_key,
                  "request_sha256": request_hash, "approval_sha256": approval, "endpoint": endpoint, "mode": args.mode,
                  "model": args.model, "redacted_request": safe_body, "sources": safe_sources, "output_policy": output_policy,
                  "quoted_usd": format(amount, "f"), "pricing": envelope["pricing"], "governance": envelope["governance"]}
        try: exclusive_record(ledger_path, record)
        except FileExistsError: raise SystemExit("Another process acquired this job ledger")
        record.update(state="create-started", create_started_at=utc_now()); save_record(ledger_path, record)
        try: created, zdr = request_json("POST", API + endpoint, api_key, body, retries=0)
        except APIProblem as exc:
            ambiguous = exc.status is None or exc.status >= 500 or exc.status in {408, 409}
            record.update(state="create-outcome-unknown" if ambiguous else "create-rejected-known", error_kind=exc.kind,
                          http_status=exc.status, error_body_sha256=exc.body_sha256, replay_allowed=False, create_outcome_at=utc_now())
            save_record(ledger_path, record)
            raise SystemExit("Create stopped; ambiguous jobs must be reconciled and no job key is replayable")
        request_id = created.get("request_id")
        if not isinstance(request_id, str) or not request_id or len(request_id) > 128:
            record.update(state="create-outcome-unknown", error_kind="unusable-success-request-id",
                          response_sha256=sha256_bytes(canonical(created)), replay_allowed=False, create_outcome_at=utc_now())
            save_record(ledger_path, record); raise SystemExit("No usable request_id returned; reconcile before replacement")
        record.update(state="request-pending", request_id=request_id, create_zdr_header=zdr, submitted_at=utc_now()); save_record(ledger_path, record)
    if args.require_zdr and zdr != "true":
        record["zdr_mismatch"] = True; save_record(ledger_path, record)
        raise SystemExit("ZDR header was not true; stop handling and escalate immediately")
    deadline = time.monotonic() + 20 * 60
    while time.monotonic() < deadline:
        try: result, poll_zdr = request_json("GET", API + "/videos/" + urllib.parse.quote(request_id, safe=""), api_key, retries=3)
        except APIProblem as exc:
            record.update(poll_error_kind=exc.kind, poll_http_status=exc.status, poll_error_body_sha256=exc.body_sha256); save_record(ledger_path, record)
            raise SystemExit("Polling failed; resume the known request ID later")
        status = result.get("status"); record.update(last_status=status, poll_zdr_header=poll_zdr, progress=result.get("progress"), last_polled_at=utc_now()); save_record(ledger_path, record)
        if args.require_zdr and poll_zdr != "true":
            record["zdr_mismatch"] = True; save_record(ledger_path, record)
            raise SystemExit("A poll response did not confirm ZDR; stop handling and escalate")
        if status == "done":
            video = result.get("video") or {}
            if video.get("respect_moderation") is not True:
                record.update(state="moderation-blocked", moderation_value=video.get("respect_moderation")); save_record(ledger_path, record)
                raise SystemExit("Output did not pass moderation; do not publish or bypass")
            url = video.get("url")
            if not url: raise RuntimeError("Done response has no video URL")
            artifact = download_mp4(url, args.output, set(output_hosts), ledger_path, record)
            ticks = (result.get("usage") or {}).get("cost_in_usd_ticks")
            if not isinstance(ticks, int) or ticks < 0: reconciliation = {"status": "missing-or-invalid-stop-further-creates"}
            else:
                actual = Decimal(ticks) / Decimal(10_000_000_000)
                reconciliation = {"status": "matched" if actual == amount else "variance-stop-further-creates",
                                  "actual_usd": format(actual, "f"), "variance_usd": format(actual - amount, "f")}
                if actual > maximum: reconciliation["status"] = "ceiling-exceeded-stop-further-creates"
            record.update(state="artifact-secured", artifact=dict(artifact, path=str(pathlib.Path(args.output).resolve())),
                          output_duration=video.get("duration"), cost_in_usd_ticks=ticks, cost_reconciliation=reconciliation,
                          result_model=result.get("model"), human_review="pending", stored_at=utc_now())
            save_record(ledger_path, record); print(json.dumps({"artifact": record["artifact"], "cost_reconciliation": reconciliation}, indent=2)); return
        if status in {"failed", "expired"}:
            record.update(state="request-" + status, error_sha256=sha256_bytes(str(result.get("error", "")).encode("utf-8"))); save_record(ledger_path, record)
            raise SystemExit(f"Video request {status}")
        if status != "pending": raise RuntimeError(f"Unknown status {status!r}")
        time.sleep(5)
    record["state"] = "local-poll-timeout"; save_record(ledger_path, record)
    raise SystemExit("Local polling ended; resume GET by request_id, never create a replacement")

if __name__ == "__main__":
    main()

ドライランの例：6 秒、720p の T2V を 1 件、0.42 USD と見積もります。キーやネットワークリクエストは使用しません。各エビデンス用プレースホルダーを保護された 64 桁の 16 進ダイジェストに置き換え、現在のタイムゾーン対応の価格確認時刻を使用してください。

python xai_video_guard.py --mode text --model grok-imagine-video --duration 6 --resolution 720p --aspect-ratio 16:9 --job-key cup-v1 --output-host vidgen.x.ai --pricing-checked-at 2026-07-10T12:00:00Z --pricing-evidence-sha256 '<64-hex>' --rights-sha256 '<64-hex>' --moderation-sha256 '<64-hex>' --qa-plan-sha256 '<64-hex>' --prompt "Wide dawn shot of a blue ceramic cup on a rain-dark cafe table. A hand lifts it while the camera makes one slow push-in. End beside the window. Quiet room tone and one soft ceramic clink." --ledger attempt-cup.json

保護されたペイロードと現在の価格を確認した後、引数を一切変更せずに同じコマンドを再実行し、--execute --approval-sha256 '<printed-digest>' --max-cost-usd 0.42 を追加します。XAI_API_KEY をエクスポートするのはサーバー側のシェル内だけにしてください。実際のキーをファイル、共有コマンド履歴、ブラウザーコード、ログ、チャットに貼り付けないでください。

1.5 の I2V ドライランの例：12 秒、1080p の出力は 3.010 USD です。ソースは使用権が確認済みである必要があり、その SHA-256 を記録します。

python xai_video_guard.py --mode image --model grok-imagine-video-1.5 --duration 12 --resolution 1080p --aspect-ratio 16:9 --image file_authorized_waterfall --source-local ./authorized-waterfall.png --source-sha256 REPLACE_WITH_LOCAL_FILE_SHA256 --job-key waterfall-v1 --output-host vidgen.x.ai --pricing-checked-at 2026-07-10T12:00:00Z --pricing-evidence-sha256 '<64-hex>' --rights-sha256 '<64-hex>' --moderation-sha256 '<64-hex>' --qa-plan-sha256 '<64-hex>' --prompt "The waterfall surges from the starting frame; mist rolls toward the lens as the camera slowly dollies back. End on the full cliff. Deep rushing water and distant birds." --ledger attempt-waterfall.json

この例では意図的に storage_options をリクエストせず、ダウンロードします。サーバー側の永続化を追加するのは、そのプライバシーのライフサイクルを決定した後にしてください。

非同期状態をエビデンスとして扱う

作成すると request_id が返されます。数秒ごとに GET /v1/videos/{request_id} をポーリングします。xAI の例では 5 秒間隔です。ステータスは厳密に pending、done、expired、failed のいずれかです。ローカル SDK のタイムアウトは、プロバイダー側の失敗を意味しません。リクエスト ID を保持し、ポーリングを再開してください。

done になったら、video.respect_moderation が true であることを検証し、model、progress、duration、整数のコストティックを取得してから、一時的な vidgen.x.ai URL を速やかにダウンロードします。通常の URL については、正確な有効期間は公表されていません。バッチの画像／動画結果 URL は別扱いで、有効期間は 1 時間です。

ffprobe を実行し、ファイル全体をデコードします。コンテナー、動画コーデック、寸法、長さ、フレームレート、すべてのオーディオストリーム／コーデック／チャンネル数／サンプルレート、バイト数、SHA-256 を記録します。冒頭フレームの忠実度、時間経過に伴う同一性、接触時の物理表現、カメラの連続性、編集時の不変条件または延長部分の継ぎ目、音声同期、発話の正確さ、望ましくないテキスト／ロゴ、モデレーション、一般公開への準備状況を確認します。

ストレージ、Files、Batchを意図的に選択する

ガバナンス対象の入力を繰り返し使用する場合は、公開ソースURLよりもプライベートなFilesのfile_idを優先してください。xAI FilesはImagineの参照用にPNG/JPEG/WebPとMP4を受け付けます。ファイルはチーム単位で管理されます。expires_afterを3,600～2,592,000秒（1時間～30日）に設定しない限り、アップロードしたファイルは削除するまで保持されます。制作記録で不要になった時点で削除してください。

Filesのストレージ料金は別途発生し、現在の料金ページでは$0.025/GiB/day、Filesのダウンロードは$0.20/GiBです。承認済みワークフローでfile_idまたはstorage_optionsによる永続化を利用する場合は、期間と使用量に応じたこのコストを必ず含めてください。安全策を講じた例では、通常の一時的な結果を使い、ガバナンス管理下でローカルにダウンロードすることで、変動するストレージ料金を回避しています。

storage_optionsを使うと、Imagineの出力をプライベートに永続化できます。

{"filename":"approved-cut.mp4","expires_after":86400,"public_url":false}

expires_afterを省略すると、ファイルは削除されるまで永久に保持されます。public_url: trueを設定すると、認証不要の共有リンクが作成されます。有効期限を設定しない場合、無期限に利用できる可能性があります。プライベートストレージを優先してください。公開配信が必要な場合は、1時間～30日の範囲で最短の有効期限を設定し、対象者と有効期限を記録して、配信後に無効化してください。公開URLがファイルの有効期限を超えることはありません。署名付きURLや推測困難なURLを通常のログに記録しないでください。

ZDRとFilesの永続化またはstorage_optionsとの互換性は、明確に文書化されていません。永続ストレージは、一時的な推論とは概念的に異なります。厳格なZDR要件がある作業では、xAIが契約上の動作を確認するまでFilesへの永続化を避け、独自にガバナンス管理下で暗号化したコピーを保持してください。

Batchは、JSONLジョブの混在を含め、/v1/videos/generations、/edits、/extensionsをサポートします。料金は通常のメディア料金が適用され、動画の割引はありません。また、ほとんどのBatch処理は24時間以内に完了します。総コストの上限を設定し、安定したcustom_idを使用し、項目ごとの権利記録を用意した、レビュー済みの一括処理に限って使用してください。返されるメディアURLの有効期限は1時間です。リアルタイムの作成処理の結果が不確かな場合に、冪等性を確保する代替手段としてBatchを使用しないでください。

モデレーションを尊重し、開示し、人々を保護する

xAIは、動画がコンテンツポリシーの審査対象であることを明示しています。respect_moderation: trueは公開に必要ですが、それだけでは十分ではありません。プロバイダーの安全策を回避したり、来歴メタデータやウォーターマークを削除したりしないでください。AUPでは、安全策の回避と埋め込み済みの来歴情報の削除が明示的に禁止されています。

プロンプト、ソース画像や動画、ブランド、場所、音楽、声、実演に関するすべての権利を取得してください。個人を特定できる人物については、肖像や声、合成による加工、編集範囲、対象者、地域、期間、商業的な文脈を対象とする、十分な説明に基づいた許諾を記録してください。AUPでは、欺瞞的ななりすまし、同意のない親密またはポルノ的な肖像の使用、中傷的な描写や誤った印象を与える描写、プライバシー権やパブリシティ権の侵害、詐欺、児童の性的搾取が禁止されています。オンラインで見つけた有名人の画像は、本人の同意を意味しません。

現行のEnterprise Termsでは、適用法に従い、xAIと顧客との間では顧客が入力の権利を保持し、出力を所有します。ただし、xAIの技術は対象外です。この権利配分によって、第三者の人物、ブランド、実演、著作権で保護されたソースに関する権利が生じるわけではありません。また、出力が独自のものであるとは限りません。Termsでは、出力を人間が生成したものと表示すること、および顧客またはそのプロバイダーのML/AIモデルのトレーニングに出力を使用することが禁止されています。

視聴者に誤解を与えるおそれがある場合は、AIによる重要な生成や改変を明確に開示し、管轄区域やプラットフォームに応じた政治、広告、生体認証、児童の安全、合成メディアに関するラベルを適用してください。元のメディア、ハッシュ、プロンプト、モデルエイリアス、xAIへのリクエストや費用の証跡、編集内容、開示に関する判断を保持してください。xAIのモデレーションが、真実性、推薦、著作権上の許諾、法令遵守を保証すると主張しないでください。

コンシューマー向けではなく、API データの契約条件を定める

API を直接利用する顧客には、xAI の Enterprise Terms、DPA、AUP が適用されます。公開されているコンシューマー向けプライバシーポリシーには、ビジネス顧客向けに処理される xAI API データは対象外であると明記されています。また、X 上の Grok には X のポリシーが適用されるとも記載されています。

xAI の API セキュリティ FAQ によると、API の入力／出力は明示的な許可がない限りトレーニングに使用されず、不正利用の監査のため通常 30 日間保持された後に削除されます。Enterprise Terms には、ユーザーコンテンツは基盤モデル／AI システムのトレーニングに使用されないこと、また、明記された法的、安全性、セキュリティ、モデレーション、調査に関する例外を条件として、やり取りから 30 日以内に削除されることが定められています。

ZDR はエンタープライズ専用で、チーム全体に適用されます。xAI Console で有効になっていることを確認し、作成時とポーリング時のレスポンスヘッダー x-zero-data-retention: true を記録してください。モデレーションは引き続き実行されます。FAQ では、データは一切永続化されず、配信後に記録も残らないとされています。一方、現行の Enterprise Terms では、永続的なログ／バックアップはその後作成されないものの、配信時または完了から 1 時間後のいずれか早い方まで、一時的な処理アーティファクトが許容されると、より正確に定められています。法的条件と注文書を優先し、この文言の違いを記録してください。また、「処理ゼロ」と宣伝したり、独自の監査／保持義務を回避したりしないでください。

現行の Enterprise Terms では、ZDR が有効な API 以外に個人データを意図的に送信しないよう、顧客に求めています。個人データは、契約上の ZDR チームを通じてのみ取り扱い、適切な DPA、適法な根拠、通知、データ最小化、移転評価、アクセス制御、削除計画を整えてください。PHI には BAA と ZDR も必要です。EU データレジデンシーはエンタープライズ向けオプションであり、api.x.ai からデフォルト設定だと推測できるものではありません。クラスター、サブプロセッサー、越境移転の仕組み、フェイルオーバーの動作を契約で確認してください。

主張を適切な分類に分ける

• 文書で確認できる事実： 現行の xAI 一次情報源に記載されたリクエストフィールド、制限、料金、ステータス、ポリシー文面、または観測されたレスポンスプロパティ。
• プロバイダーの主張： 品質、物理法則、指示への追従、保持、音声、レイテンシー、ベンチマークでの評価に関する xAI の説明。
• 本番環境でのヒューリスティック： 映画制作とリスク管理から導き出された、プロンプトの構成、参照素材の削減、カメラの簡素化、レビュー戦略、連続性の維持に関する実践。
• 不明： 現行のスキーマと契約で確定できない事項――FPS、エフェメラル URL の正確な有効期間、音声の仕様／制御、シード、ネガティブプロンプト、Base64／公開入力のサイズ、ZDR と Files の正確な連携、またはエンタープライズ設定なしでのデータレジデンシーなど。

重大な矛盾がある場合は、より新しいエンドポイント固有のリファレンス、xAI Console、署名済み契約を優先してください。それ以外の場合は作業を中断し、xAI サポートに問い合わせてください。2026-07-10 時点で確認されている矛盾は次のとおりです。

• 公開されているレート制限／モデルのレンダリング内容では、1 RPS、10 RPS、70 RPM の間で記載が一致していません。
• Files ガイドでは 48 MB、アップロードリファレンスでは 50 MB とされています。
• ZDR FAQ では配信後に何も永続化されないとされていますが、Enterprise Terms では配信時または 1 時間後のいずれか早い方までの一時的な境界が明記されています。
• xAI のローンチ／ニュースページではネイティブ音声について説明されていますが、REST リファレンスには音声の制御項目も技術仕様もありません。
• デフォルトの Imagine 出力 URL は単に「一時的」とされており、1 時間の有効期間が明記されているのは Batch の結果のみです。

公式情報の根拠一覧

• 動画ガイドと REST リファレンス：https://docs.x.ai/developers/model-capabilities/video/generation, https://docs.x.ai/developers/rest-api-reference/inference/videos
• I2V、参照素材、編集、延長：https://docs.x.ai/developers/model-capabilities/video/image-to-video, https://docs.x.ai/developers/model-capabilities/video/reference-to-video, https://docs.x.ai/developers/model-capabilities/video/editing, https://docs.x.ai/developers/model-capabilities/video/extension
• モデルページと料金：https://docs.x.ai/developers/models/grok-imagine-video, https://docs.x.ai/developers/models/grok-imagine-video-1.5, https://docs.x.ai/developers/pricing
• モデルの確認、コスト、制限、エラー、リリース：https://docs.x.ai/developers/rest-api-reference/inference/models, https://docs.x.ai/developers/cost-tracking, https://docs.x.ai/developers/rate-limits, https://docs.x.ai/developers/debugging, https://docs.x.ai/developers/release-notes
• ファイルと Batch：https://docs.x.ai/developers/model-capabilities/imagine/files, https://docs.x.ai/developers/model-capabilities/imagine/files/inputs, https://docs.x.ai/developers/model-capabilities/imagine/files/outputs, https://docs.x.ai/developers/files/managing-files, https://docs.x.ai/developers/files/public-urls, https://docs.x.ai/developers/advanced-api-usage/batch-api
• API セキュリティと法的情報：https://docs.x.ai/developers/faq/security, https://x.ai/legal/terms-of-service-enterprise, https://x.ai/legal/data-processing-addendum, https://x.ai/legal/acceptable-use-policy
• コンシューマー向けサービスと X の区分：https://docs.x.ai/grok/overview, https://x.ai/legal/terms-of-service, https://x.ai/legal/privacy-policy
• ネイティブ音声に関するプロバイダーの主張と 1.5 のリリース：https://x.ai/news/grok-imagine-api, https://x.ai/news/grok-imagine-video-1-5
• サービスの稼働状況：https://status.x.ai/

現行の xAI ドキュメントおよび調査インデックスでは、Grok Imagine Video 専用の公式技術モデルカードまたは安全性カードは見つかりませんでした。xAI が公式資料を公開しない限り、アーキテクチャ、学習コーパス、人口統計上の性能、ウォーターマークの実装、詳細な安全性評価、ベンチマークの汎化性能は不明として扱ってください。