runway-video

Runway Video

このスキルは、ドキュメントに記載されている https://api.dev.runwayml.com の Runway Developer API とのサーバーサイド連携に使用します。Runway ネイティブの動画モデルを対象とし、パートナーモデル、静止画の生成、一般向けウェブアプリ、Recipes、Workflows、Characters、サードパーティ製ゲートウェイは明示的に対象外とします。

調査基準日：2026-07-10。

エビデンスの表記

このスキルの記述は、次のように解釈してください。

• Fact — Runway の一次情報、または公式 OpenAPI から生成された SDK に記載されている内容です。
• Claim — Runway のマーケティング上またはポリシー上の表現であり、プランや契約によって異なる場合があります。
• Heuristic — 事実に基づく運用上の推奨事項です。デプロイ環境で検証してください。
• Unknown — 確認した Runway の一次公開資料では確立されていない内容です。暗黙に補完しないでください。

変動しやすい事実が費用、安全性、ペイロードの有効性、またはデータガバナンスに影響する場合は、本番環境で使用する前に、引用されている一次情報のページを再確認してください。

スコープ、同意、実行モードから始める

リクエストを設計する前に、次のすべてを確認してください。

1. 希望するモード：テキストから動画、画像から動画、動画編集、またはパフォーマンス転送。
2. Runway ネイティブモデルが必須か、パートナーモデルでもよいか。
3. 出力の長さ、寸法、納品形式、音声の要件、反復回数の上限。
4. プロンプト、画像、動画、顔、音声、パフォーマンス、マーク、音楽のすべての入力に対する権利。
5. 映っている人、または音声が聞こえる人それぞれからの明示的な許可。Runway の Usage Policy では、他者の画像、動画、音声を許可なく使用することを禁止しています。
6. 現実的な合成メディアによって視聴者が誤解する可能性がある場合の、開示および来歴の計画。
7. データの分類、承認済みのリージョン／サブプロセッサー、保持要件、および組織の契約に実際にトレーニング禁止やゼロデータ保持の条項が含まれているかどうか。
8. 次のいずれかの実行モード：
   • Plan/dry-run — デフォルト。検証、価格計算、承認用ダイジェストの出力を行います。有料 API の作成処理は実行しません。
   • Approved create — 正確なリクエストダイジェスト、上限付きの最大費用、明示的な環境ゲートが一致した場合に限り実行します。
   • Resume — 記録済みのタスク ID をポーリングします。代替のタスクは作成しません。
   • Cancel/delete — 個別の承認を必要とする、独立した破壊的操作です。タイムアウトから推測して実行しないでください。

必要な権利や同意がないリクエスト、モデレーションの回避、なりすまし、詐欺、同意のない親密な画像、未成年者が関わる性的コンテンツ、その他の禁止カテゴリに該当するリクエストは拒否してください。ブロックされたコンテンツを難読化して「修正」しないでください。

リクエストの振り分け

現行の Runway ネイティブ動画 API

モデル	ドキュメント記載の API モード	エンドポイント	必須の要素	現行の公開価格
gen4.5	テキスト → 動画	POST /v1/text_to_video	promptText、ratio、整数の duration 2〜10	12 credits/s
gen4.5	最初の画像 + テキスト → 動画	POST /v1/image_to_video	promptImage、promptText、ratio、整数の duration 2〜10	12 credits/s
gen4_turbo	最初の画像（+ 任意のモーションテキスト）→ 動画	POST /v1/image_to_video	promptImage、ratio、長さ 2〜10	5 credits/s
aleph2	既存の動画 + 任意のテキスト／キーフレーム → 編集済み動画	POST /v1/video_to_video	videoUri、時間指定付きキーフレーム最大5個	28 credits/s、最低56 credits
act_two	キャラクター画像／動画 + 動作参照動画 → 動画	POST /v1/character_performance	character、reference、モデル act_two	5 credits/s。公開されている Act-Two ガイドでは最低3秒と記載

Facts： gen4.5、gen4_turbo、aleph2、act_two は、現行の API モデルカタログに掲載されています。gen4_aleph は非推奨で、2026-07-30 に提供終了予定です。新しい作業で使用しないでください。

ネイティブモードの詳細

• Gen-4.5 テキストから動画生成： 比率は 1280:720 と 720:1280 です。リクエストスキーマでは、1,000 UTF-16 コード単位以下の空でないプロンプト、比率、および 2 から 10 までの整数の長さが必須です。
• Gen-4.5 画像から動画生成： 比率は 1280:720、720:1280、1104:832、832:1104、1584:672、960:960 です。現在の SDK スキーマでは、URI で指定するか、position: "first" を含む配列項目として指定する、最初のフレーム画像を 1 枚受け付けます。
• Gen-4 Turbo： 画像入力が必須です。上記 6 種類の比率に対応し、長さは 2〜10 秒の任意の値を指定できます。ネイティブの画像駆動型オプションの中では、最も低コストです。
• Aleph 2.0： 入力動画は 2〜30 秒、最大 30 fps、1080p 以下である必要があります。出力では、最大 1080p まで入力解像度が維持されます。API では、省略可能な promptText、seed、contentModeration、targetAspectRatio と、最大 5 個のキーフレームを受け付けます。各キーフレームには URI と、絶対値の seconds または正規化された at のいずれかを指定します。編集範囲を指定する場合は、すべてのキーフレームで範囲を指定するか、どのキーフレームでも指定しない必要があります。
• Act-Two： character は {type: "image"|"video", uri}、reference は {type: "video", uri} です。駆動パフォーマンスの長さは 3〜30 秒です。API では、bodyControl、1〜5 の expressionIntensity、ratio、seed、contentModeration も受け付けます。身体やジェスチャーの転送はキャラクター画像の入力を想定しています。キャラクター動画では、元の環境やカメラの動きが維持され、同じジェスチャー制御機能は利用できません。

文書間の不一致： Inputs ページには Gen-4.5 の縦向き比率 672:1584 も記載されていますが、確認した現在の OpenAPI 生成 SDK 型には含まれていません。この比率は未解決として扱い、推測で判断しないでください。上記の一般的な値を使うか、672:1584 を承認する前に最新の API リファレンスまたは SDK を再確認してください。

パートナーモデルは別途判断してください

Runway では、パートナー動画モデルとして seedance2、seedance2_fast、seedance2_mini、veo3、veo3.1、veo3.1_fast、happyhorse_1_0、gemini_omni_flash も提供しています。Magnific による動画のアップスケーリングもパートナー機能です。これらはスキーマ、料金、音声の挙動、モデレーションの診断情報、障害モード、サブプロセッサー、契約上の管理策が異なる可能性があります。

ネイティブモデルを、これらのいずれかに黙って置き換えないでください。ユーザーがパートナーモデルを選択した場合は、そのモデルに関する現在の Runway 公式スキーマと、プロバイダー固有の権利およびデータ条件を改めて調査し、新しい料金計算式を作成して、新たな承認ダイジェストを取得してください。Runway のエンタープライズ FAQ には、エンタープライズ規約と DPA がパートナーモデルにも適用され、プロバイダーが学習に使用しないことを約束していると記載されています。しかし、すべてのセルフサービス API アカウントにも同じ規約が適用される証明にはなりません。

明示的に対象外

• 静止画のルート（/v1/text_to_image）は動画生成ではなく、ステージング用の選択肢です。静止画の呼び出しは別途課金され、専用のスキル／ワークフローと承認が必要です。
• コンシューマー向け Web アプリ、Explore Mode、Apps、Edit Studio UI、Web クレジットは API 契約ではありません。Web アプリのクレジットと Developer API のクレジットは別です。
• Recipes、公開済み Workflows、リアルタイム Characters、アバター動画、音声専用エンドポイント、動画のアップスケーリングは、それぞれ別の製品／エンドポイントです。
• Replicate、fal、ComfyUI ラッパー、リバースエンジニアリングされたエンドポイント、ブラウザー自動化、その他のゲートウェイは Runway Developer API ではありません。

プロンプトを作成する前にショット仕様をまとめる

簡潔なショット仕様を記録します。

intent: "one continuous 6-second product reveal"
mode: "gen4.5 text-to-video"
subject: "matte cobalt travel mug on pale limestone"
action: "a bead of condensation rolls down the mug"
environment: "quiet studio, soft morning window light"
camera: "slow 20-degree dolly arc, medium close-up"
timing: "hold composition for first second, then arc"
style: "natural commercial cinematography, restrained contrast"
ratio: "1280:720"
duration_seconds: 6
audio: "none requested; add licensed sound in post"
continuity_anchors: ["cobalt color", "logo orientation", "limestone texture"]
acceptance: ["single shot", "logo not distorted", "no camera cut"]

次に、直接的で肯定的なプロンプトを書きます。Gen-4.5 のテキストから動画を生成する場合は、何が見えるかと、どのように動くかの両方を説明します。画像から動画を生成する場合、画像によって被写体、構図、色、照明、スタイルがすでに定まっているため、被写体、カメラ、環境の動きを強調します。反復調整では、一度に1つの変数を追加します。

ヒューリスティック：

• 短いクリップでは、1つのショットと1つの主要なアクションを優先します。
• 具体的なカメラ用語を使います。「カメラ固定」「ゆっくりドリーイン」「手持ちで追跡」「俯瞰の静止フレーム」などです。
• 否定表現を肯定的な結果に置き換えます。「カメラを動かさない」ではなく「カメラ固定」とします。
• 重要な法的文言やロゴを確実に描画するようモデルに求めないでください。承認済みのテキストやマークは、ポストプロダクションで合成します。
• プロンプトのバリエーションを比較する場合に限り、シードを固定します。同じシードを使っても類似性の補助にしかならず、決定性は保証されません。

Gen-4.5 のテキストプロンプトの例：

淡い石灰岩の上に、つや消しコバルトブルーのトラベルマグが置かれています。柔らかな朝の窓辺の光の中、結露の水滴が側面をゆっくりと流れ落ちます。カメラは、1回の連続したミディアムクローズアップで、控えめに20度のドリーアークを描きます。自然なコマーシャルシネマトグラフィー、くっきりとした製品のシルエット、背景は穏やかにぼけています。

Aleph では、動作を表す動詞と対象を絞った変化を使います。「俳優、衣装、カメラの動き、セットのレイアウトを保ちながら、部屋を冷たい月明かりで照らし直してください。」まずは最小限の編集から始めます。正確な見た目が重要な場合はキーフレームを使い、キーフレームでは見えない動きや変化には promptText を使います。

Act-Two では、品質を主に左右するのは文章ではなくリファレンスです。被写体は1人にし、顔が常に見えるようにします。腰から上、またはそれより近い画角にし、照明を均一に保ち、カットを入れず、ボディコントロールを有効にする場合は開始時に手が見えるようにし、動きは自然にします。まずは expressionIntensity: 3 から始めます。値を上げると表情やアーティファクトが増えることがあり、下げると安定性が向上する場合があります。

API とアセットの契約

すべての API リクエストで、次を使用します。

Base URL: https://api.dev.runwayml.com
Authorization: Bearer $RUNWAYML_API_SECRET
X-Runway-Version: 2024-11-06
Content-Type: application/json

シークレットはサーバー側に保持します。ソース、ログ、プロンプト、ブラウザーコード、アセット URL、承認記録、出力のダウンロードリクエストに決して含めないでください。API のベース URL は固定し、エンドユーザーから受け取らないでください。

入力メディア

Runway は HTTPS URL、データ URI、runway:// の一時アップロード URI を受け付けます。

入力	HTTPS URL	データ URI	一時アップロード
画像	16 MB	エンコード済みで 5 MB	200 MB
動画	32 MB	エンコード済みで 16 MB	200 MB
音声	32 MB	エンコード済みで 16 MB	200 MB

HTTPS 入力では、IP アドレスではなくホスト名を使用し、HEAD に応答し、正確な Content-Type と Content-Length を返し、リダイレクトなしで 200 を返す必要があります。Runway のフェッチタイムアウトは10秒と記載されています。対応する画像エンコーディングは JPEG、PNG、WebP です。GIF はサポートされていません。対応する動画コンテナー／コーデックについては、Inputs ページに記載されています。

ヒューリスティック： 承認済みのオブジェクトストアがすでにある場合は、プライベートな完全一致ホスト許可リストと、有効期間の短い署名付きオブジェクト URL を使用します。それ以外の場合は、公式の一時アップロード API を使用します。アップロード前にコンテンツをローカルで検証し、その SHA-256 を承認記録にひも付けます。エンドユーザーに任意のフェッチ URL を入力させないでください。Runway 自体が URL に制限を適用していても、アプリケーション内で SSRF やデータ流出の攻撃対象になり得ます。

runway:// の一時アップロードは24時間で期限切れになり、レート制限があり、購入済みクレジットが必要で、サイズは512バイト〜200 MBです。マルチパートアップロードに失敗した場合、再試行せず、新しいアップロードセッションを作成してください。アップロードは状態を持つため、API シークレットやソースバイトをログに含めずに URI を記録してください。

出力メディア

タスクが成功すると、1 つ以上の URL を含む output 配列が返されます。Runway によると、これらの URL は 24～48 時間以内に期限切れとなるため、永続ストレージにダウンロードする必要があります。製品 URL として公開しないでください。公式 SDK の型定義には、タスクを再取得すると新しい URL が返される場合があると記載されています。

出力 URL は、信頼できない署名付き URL として扱ってください。

• これらの URL に Authorization を付けないでください。
• HTTPS、ユーザー情報なし、デフォルトポート、パブリック DNS、出力ホストの厳密な許可リスト、各ホップで再検証するリダイレクト数の上限、バイト数の上限、想定するメディアタイプ、およびコンテナのシグネチャを必須としてください。
• 非公開の一時ファイルにダウンロードし、SHA-256 を計算し、ffprobe を実行してから、ffmpeg -v error -i … -f null - で最後までデコードし、その後アトミックに公開してください。
• タスク ID、リクエストダイジェスト、モデル、エンドポイント、プロンプトのハッシュまたは保護されたプロンプト、ソースのハッシュ、承認、タイムスタンプ、出力ハッシュ、コーデック／寸法／fps／長さ／音声ストリームの有無、および開示・権利に関する記録を保存してください。セキュリティレビュー済みのインシデント対応ワークフローで必要な場合を除き、署名付きクエリ文字列は保存しないでください。

公開 API の出力ページでは、現在のすべてのモデルで単一のコンテナ／コーデックが使われるとは保証されていません。MP4 のみをサポートするバリデーターは、他の出力タイプを別名で保存せず、フェイルクローズしてください。

作成前に料金を確認する

Developer credits は、適用される税を加える前の価格が 1 クレジットあたり $0.01 と公開されています。

ネイティブモードでは、次の計算式を使用してください。

gen4.5:      credits = 12 × requested_output_seconds
gen4_turbo:  credits =  5 × requested_output_seconds
aleph2:      credits_upper_bound = max(56, 28 × ceil(input_video_seconds))
act_two:     credits_upper_bound = 5 × max(3, ceil(driving_performance_seconds))
usd_upper_bound = credits_upper_bound × 0.01

Aleph と Act-Two の上限計算式は、ソースの長さが端数の場合の承認に使用する保守的なヒューリスティックです。端数秒数に対する API の丸め動作は、公開資料では明らかにされていません。Aleph のコンシューマー向け Edit Studio では、範囲指定編集ではその範囲分のクレジットのみを使用すると説明されていますが、確認済みの API 料金ページには API の範囲課金ルールが記載されていません。Developer API のドキュメントまたは契約で別途確認できない限り、ソース全体の長さに基づいて料金を見積もってください。

例：

リクエスト	クレジット	税抜 USD
Gen-4.5、6 s	72	$0.72
Gen-4 Turbo、6 s	30	$0.30
Aleph 2.0、入力 5 s	140	$1.40
Act-Two、パフォーマンス 8 s	40	$0.40

モデレーションされた生成も、成功した生成と同じ料金がかかります。モデレーションの事前確認に有料の Runway 呼び出しを使用しないでください。アプリケーション内で事前にモデレーションを行い、モデレーションの失敗も予算に含めてください。

Runway の公開 API ドキュメントには、ティアに応じた同時実行数、24 時間単位のローリング生成上限、および 30 日間の支出上限が記載されています。同時実行数を超えると、タスクは THROTTLED となり、キューに残ります。1 日あたりの作成上限を超えると、作成時に 429 が返されます。一般的な 1 分あたりのリクエスト上限は公開されていませんが、これは独自のキュー、予算、悪用防止策なしに作成してよいという意味ではありません。

有料リクエストの厳密な承認

有料リクエストは、作成時点で次の条件をすべて満たす必要があります。

1. ドライランにより、エンドポイント、モデル、正確な JSON ペイロード、入力の SHA-256 値、ローカルジョブキー、出力ポリシー、現在の価格情報の日付、および保守的に見積もった最大 USD 費用を含む正規リクエストエンベロープが生成されていること。
2. 保護されたインターフェースで、人間がリクエストの非マスキング版を確認していること。
3. 人間が「Runway を使う」だけではなく、そのエンベロープの正確なダイジェストを承認していること。
4. SEND_RUNWAY_REQUEST=1 が設定されていること。
5. APPROVE_RUNWAY_PAID_CALL=YES が設定されていること。
6. RUNWAY_APPROVAL_SHA256 が表示された SHA-256 と完全に一致すること。
7. RUNWAY_MAX_COST_USD が設定されており、計算された上限以上であること。
8. POST の前に、作成台帳がジョブキーを排他的に取得していること。

プロンプト、モデル、エンドポイント、シード、長さ、比率、参照、コンテンツモデレーション設定、価格表、またはソースハッシュに変更があれば、承認は無効になります。

重複作成の制御と非同期ライフサイクル

Runway の作成処理はタスク ID を返します。確認済みの API リファレンスと OpenAPI から生成された SDK には、これらの作成エンドポイント用の冪等性キーのパラメーターは記載されていません。そのため、次のようにしてください。

• POST の前に、排他的作成セマンティクスを使ってローカルのジョブレコードを作成します。
• リクエストダイジェストと state: creating を先に保存します。
• 作成 POST はそれぞれ 1 回だけ送信します。作成呼び出しの自動再試行は無効にします。
• 返されたタスク ID は、ただちに永続的に保存します。
• クライアントがタイムアウトする、切断する、クラッシュする、形式不正の成功応答を受け取る、またはタスク ID の保存前にバイト列が送信された可能性がある状態で 5xx を受け取った場合、ジョブを曖昧としてマークします。自動的に再作成しないでください。プロバイダーのサポート窓口や利用記録を通じて照合するか、人の承認のもと、新しいジョブキーで重複の可能性を受け入れてください。
• タスク ID が判明している場合は、GET /v1/tasks/{id} を再開します。ポーリングがタイムアウトしたという理由だけで、新しい生成に置き換えてはいけません。

公式 SDK は、デフォルトで接続エラー、408、409、429、5xx を 2 回再試行します。読み取りには便利ですが、プロバイダーによる冪等性の保証がない、有料の作成処理には安全ではありません。SDK を使用する場合は、作成用クライアントを max_retries=0 でインスタンス化します。読み取り用の別クライアントでは、安全な GET を再試行できます。

ポーリングは 5 秒より短い間隔で行わないでください。再試行可能な読み取りエラーには、ジッターと指数バックオフを適用します。ステータスは、次のものだけを認識してください。

ステータス	意味	対応
PENDING	受け付け済み、待機中	ポーリングを続ける
THROTTLED	保存済みだが、同時実行制限によりキューに追加されていない	ポーリングを続ける。再作成しない
RUNNING	実行中。進捗情報が含まれる場合がある	ポーリングを続ける
SUCCEEDED	終了状態。出力 URL を利用可能	ただちにダウンロードして検証する
FAILED	終了状態。failureCode とサニタイズ済みの failure を確認する	むやみに再試行しない
CANCELLED	終了状態	停止する

公式 SDK スキーマでは、CANCELLED の「l」は 2 つです。

ポーリング期限や SDK の待機タイムアウトによって、タスクがキャンセルされることはありません。DELETE /v1/tasks/{id} は、PENDING、THROTTLED、RUNNING のタスクをキャンセルする一方、それ以外のタスクとその永続化された出力データを削除するという、複数の用途に使われます。読み取り後に DELETE を呼び出す方法では、2 つの呼び出しの間にタスクが完了する可能性があり、競合が発生します。DELETE を自動的に呼び出してはいけません。終了状態のタスクが削除される可能性を了承したうえで、別途、明示的な破壊的操作の承認を必須にしてください。キャンセル時の返金・課金ルールは公開ドキュメントに記載されていないため、キャンセルはコスト抑制の保証にはなりません。

本番環境向け完全リファレンス

以下の標準ライブラリのみを使用する Python クライアントは、Gen-4.5 のテキストから動画への変換を直接実装し、ネイティブの画像から動画への変換、Aleph、または Act-Two 用に検証済みのソースプランを受け付けます。デフォルトではドライランで動作し、有料ペイロード、ソース証拠、価格、出力先ホスト／ネットワークポリシー、ガバナンス証拠に厳密にひも付けた承認を適用し、排他的なローカル台帳を使用します。作成処理は再試行せず、ポーリングは安全に再試行し、自動キャンセルを拒否します。また、API ベアラートークンを使わずにダウンロードし、ffprobe と ffmpeg による全体のデコードで MP4 を検証します。

Python 3.11 以降、ffprobe、ffmpeg が必要です。ペイロードのテンプレートの後にある付属のソースプランビルダーを使うと、このライフサイクルを弱めることなく、ほかのネイティブモードを利用できます。

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

import hashlib
import http.client
import ipaddress
import json
import os
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
import uuid
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation, ROUND_CEILING
from pathlib import Path

API_ORIGIN = "https://api.dev.runwayml.com"
API_VERSION = "2024-11-06"
CREATE_PATH = "/v1/text_to_video"
PRICE_CREDITS_PER_SECOND = 12
USD_PER_CREDIT = Decimal("0.01")
NONTERMINAL = {"PENDING", "THROTTLED", "RUNNING"}
TERMINAL = {"SUCCEEDED", "FAILED", "CANCELLED"}
MAX_JSON = 256 * 1024
MAX_VIDEO = 512 * 1024 * 1024
MAX_TOOL_OUTPUT = 512 * 1024
UNRESOLVED_GOVERNANCE = [
    "training-use-controlling-terms", "retention-and-deletion-SLA", "zero-data-retention",
    "data-residency", "human-review", "partner-subprocessors", "incident-notification",
]
LEDGER_KEYS = {
    "schemaVersion", "state", "createdAt", "requestHash", "approvalSha256", "model", "endpoint",
    "apiVersion", "promptSha256", "sources", "outputPolicy", "evidence", "price",
    "unresolvedGovernance", "taskId", "submittedAt", "createOutcomeAt", "errorKind", "httpStatus",
    "errorBodySha256", "replayAllowed", "responseSha256", "terminalAt", "failureCode", "artifactError",
    "outputUrlSha256", "stageName", "downloadStartedAt", "artifactStaged", "artifactValidatedAt",
    "storedAt", "artifact", "manifestSha256", "costReconciliation",
}


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


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


NO_REDIRECT = urllib.request.build_opener(NoRedirect)


def die(message: str) -> "NoReturn":
    raise SystemExit(message)


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


def sha256_bytes(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


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


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


def atomic_json(path: Path, value: object) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, temp_name = tempfile.mkstemp(prefix=".ledger-", suffix=".json", dir=path.parent)
    try:
        os.fchmod(fd, 0o600)
        with os.fdopen(fd, "wb") as stream:
            stream.write(canonical(value) + b"\n")
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temp_name, path)
        fsync_directory(path.parent)
    finally:
        if os.path.exists(temp_name):
            os.unlink(temp_name)


def exclusive_ledger(path: Path, initial: object) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY
    fd = os.open(path, flags, 0o600)
    with os.fdopen(fd, "wb") as stream:
        stream.write(canonical(initial) + b"\n")
        stream.flush()
        os.fsync(stream.fileno())


def read_json(path: Path) -> dict:
    raw = path.read_bytes()
    if len(raw) > MAX_JSON:
        die("ledger exceeds JSON cap")
    value = json.loads(raw)
    if not isinstance(value, dict):
        die("ledger is not an object")
    return value


def exact_hosts(raw: str, name: str) -> tuple[str, ...]:
    result = []
    for item in raw.split(","):
        host = item.strip().rstrip(".").lower()
        if not host:
            continue
        if "*" in host or "/" in host or ":" in host:
            die(f"{name} accepts exact DNS hostnames only")
        try:
            ipaddress.ip_address(host)
        except ValueError:
            pass
        else:
            die(f"{name} rejects IP literals")
        result.append(host)
    if not result:
        die(f"{name} must contain at least one reviewed exact hostname")
    return tuple(sorted(set(result)))


def validate_public_https(url: str, allowed_hosts: set[str]) -> tuple[str, str, list[str]]:
    parsed = urllib.parse.urlsplit(url)
    if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password or parsed.fragment:
        die("output URL must be HTTPS with a hostname and no userinfo")
    if parsed.port not in (None, 443):
        die("output URL must use default HTTPS port")
    host = parsed.hostname.rstrip(".").lower()
    try:
        ipaddress.ip_address(host)
    except ValueError:
        pass
    else:
        die("output URL must not use an IP literal")
    if host not in allowed_hosts:
        die(f"output host {host!r} is not in RUNWAY_OUTPUT_HOSTS; resume after review")
    addresses = []
    for item in socket.getaddrinfo(host, 443, type=socket.SOCK_STREAM):
        address = ipaddress.ip_address(item[4][0])
        if not address.is_global:
            die(f"output host resolves to non-public address {address}")
        addresses.append(str(address))
    if not addresses:
        die("output host did not resolve")
    target = parsed.path or "/"
    if parsed.query:
        target += "?" + parsed.query
    return host, target, sorted(set(addresses))


class PinnedHTTPSConnection(http.client.HTTPSConnection):
    def __init__(self, host: str, address: str, timeout: float):
        super().__init__(host, 443, timeout=timeout, context=ssl.create_default_context())
        self.address = address

    def connect(self) -> None:
        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 bounded_process(argv: list[str], timeout: int, cap: int = MAX_TOOL_OUTPUT) -> bytes:
    process = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    chunks: list[bytes] = []
    total = 0
    overflow = threading.Event()
    lock = threading.Lock()

    def drain(stream) -> None:
        nonlocal total
        while True:
            data = stream.read(65536)
            if not data:
                return
            with lock:
                total += len(data)
                if total > 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()
        die(f"tool timeout: {Path(argv[0]).name}")
    for thread in threads:
        thread.join(timeout=2)
    if overflow.is_set():
        die(f"tool output exceeds cap: {Path(argv[0]).name}")
    if code != 0:
        die(f"tool failed: {Path(argv[0]).name}; diagnosticsSha256={sha256_bytes(b''.join(chunks))}")
    return b"".join(chunks)


def api_json(method: str, path: str, body: dict | None, retry_reads: bool) -> dict:
    secret = os.environ.get("RUNWAYML_API_SECRET", "")
    if not secret:
        die("RUNWAYML_API_SECRET is required for network operations")
    url = API_ORIGIN + path
    headers = {
        "Authorization": f"Bearer {secret}",
        "Accept": "application/json",
        "X-Runway-Version": API_VERSION,
    }
    data = None
    if body is not None:
        data = canonical(body)
        headers["Content-Type"] = "application/json"
    attempts = 5 if retry_reads else 1
    for attempt in range(attempts):
        request = urllib.request.Request(url, data=data, headers=headers, method=method)
        try:
            with NO_REDIRECT.open(request, timeout=30) as response:
                raw = response.read(MAX_JSON + 1)
                if len(raw) > MAX_JSON:
                    raise APIProblem("response-too-large")
                try:
                    value = json.loads(raw or b"{}")
                except json.JSONDecodeError:
                    raise APIProblem("invalid-json", response.status, sha256_bytes(raw))
                if not isinstance(value, dict):
                    raise APIProblem("non-object-json", response.status, sha256_bytes(raw))
                return value
        except urllib.error.HTTPError as exc:
            error_body = exc.read(MAX_JSON + 1)
            body_hash = sha256_bytes(error_body)
            retryable = retry_reads and exc.code in {429, 502, 503, 504}
            if not retryable or attempt + 1 == attempts:
                raise APIProblem("http-error", exc.code, body_hash)
        except APIProblem:
            raise
        except (urllib.error.URLError, TimeoutError, OSError) as exc:
            if not retry_reads:
                raise APIProblem("transport-" + type(exc).__name__)
            if attempt + 1 == attempts:
                raise APIProblem("poll-transport-" + type(exc).__name__)
        time.sleep(min(30.0, 2 ** attempt) * (1.0 + random.random() * 0.5))
    die("unreachable")


def validate_task_id(value: object) -> str:
    if not isinstance(value, str):
        die("missing task id")
    try:
        uuid.UUID(value)
    except ValueError:
        die("task id is not a UUID")
    return value


def evidence_sha(name: str) -> str:
    value = os.environ.get(name, "").strip().lower()
    if len(value) != 64 or any(character not in "0123456789abcdef" for character in value):
        die(f"{name} must be the SHA-256 of the reviewed protected evidence")
    return value


def finite_positive_decimal(name: str) -> Decimal:
    try:
        value = Decimal(os.environ.get(name, ""))
    except InvalidOperation:
        die(f"{name} must be a decimal")
    if not value.is_finite() or value <= 0:
        die(f"{name} must be finite and positive")
    return value


def file_sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def sanitized_sources(sources: list[dict]) -> list[dict]:
    return [{key: value for key, value in item.items() if key not in {"localPath", "uri"}} for item in sources]


def redacted_payload(payload: dict) -> dict:
    def redact(value, key=""):
        if key in {"uri", "videoUri", "promptImage"} and isinstance(value, str):
            return "sha256:" + sha256_bytes(value.encode("utf-8"))
        if key == "promptText" and isinstance(value, str):
            return "sha256:" + sha256_bytes(value.encode("utf-8"))
        if isinstance(value, dict):
            return {child_key: redact(child, child_key) for child_key, child in value.items()}
        if isinstance(value, list):
            return [redact(child, key) for child in value]
        return value
    return redact(payload)


def reprobe_source(path: Path, kind: str) -> dict:
    with path.open("rb") as stream:
        header = stream.read(32)
    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")
    ):
        die("source image signature is not JPEG, PNG, or WebP")
    raw = bounded_process(
        ["ffprobe", "-v", "error", "-show_streams", "-show_format", "-of", "json", str(path)], 30, MAX_JSON
    )
    try:
        metadata = json.loads(raw)
        video = next(item for item in metadata["streams"] if item.get("codec_type") == "video")
        result = {"width": int(video["width"]), "height": int(video["height"])}
        if kind == "video":
            duration = Decimal(str(metadata["format"]["duration"]))
            numerator, denominator = str(video["avg_frame_rate"]).split("/", 1)
            fps = Decimal(numerator) / Decimal(denominator)
            if not duration.is_finite() or duration <= 0 or not fps.is_finite() or fps <= 0:
                die("re-probed source timing is invalid")
            result.update({"durationSeconds": duration, "fps": fps})
    except (KeyError, StopIteration, TypeError, ValueError, InvalidOperation, ZeroDivisionError, json.JSONDecodeError):
        die("source probe metadata is unusable")
    bounded_process(["ffmpeg", "-nostdin", "-v", "error", "-i", str(path), "-f", "null", "-"], 300)
    return result


def load_source_plan(path: Path) -> tuple[str, str, dict, list[dict], Decimal]:
    plan = read_json(path)
    if set(plan) != {"schemaVersion", "model", "endpoint", "payload", "sources", "billingSeconds"}:
        die("source plan has unexpected or missing top-level fields")
    model = plan.get("model")
    endpoint = plan.get("endpoint")
    expected = {
        "gen4.5": "/v1/image_to_video", "gen4_turbo": "/v1/image_to_video",
        "aleph2": "/v1/video_to_video", "act_two": "/v1/character_performance",
    }
    if model not in expected or endpoint != expected[model]:
        die("source plan model/endpoint mismatch")
    payload, sources = plan.get("payload"), plan.get("sources")
    if not isinstance(payload, dict) or payload.get("model") != model or not isinstance(sources, list) or not sources:
        die("source plan payload/sources are invalid")
    if model in {"gen4.5", "gen4_turbo"}:
        image = payload.get("promptImage")
        uris = [image] if isinstance(image, str) else [item.get("uri") for item in image or [] if isinstance(item, dict)]
    elif model == "aleph2":
        uris = [payload.get("videoUri")] + [item.get("uri") for item in payload.get("keyframes", []) if isinstance(item, dict)]
    else:
        uris = [payload.get("character", {}).get("uri"), payload.get("reference", {}).get("uri")]
    if len(uris) != len(sources) or any(not isinstance(uri, str) for uri in uris):
        die("payload URI/source evidence cardinality mismatch")
    allowed_source_keys = {
        "role", "kind", "transport", "localPath", "bytes", "sha256", "uriSha256",
        "contentType", "width", "height", "fps", "durationSeconds", "uploadReceiptSha256",
        "approvedHost", "validatedAt", "uploadState",
    }
    for uri, source in zip(uris, sources):
        if not isinstance(source, dict) or not set(source).issubset(allowed_source_keys):
            die("source evidence contains a non-allowlisted field")
        local = Path(source.get("localPath", "")).resolve()
        if not local.is_file() or local.stat().st_size != source.get("bytes") or file_sha256(local) != source.get("sha256"):
            die("source changed after validation; generate a new plan and approval")
        if source.get("kind") not in {"image", "video"}:
            die("source kind is invalid")
        fresh = reprobe_source(local, source["kind"])
        if fresh["width"] != source.get("width") or fresh["height"] != source.get("height"):
            die("source dimensions differ from plan evidence")
        if source["kind"] == "video":
            try:
                recorded_duration = Decimal(str(source.get("durationSeconds")))
                recorded_fps = Decimal(str(source.get("fps")))
            except InvalidOperation:
                die("recorded source timing is invalid")
            if fresh["durationSeconds"] != recorded_duration or fresh["fps"] != recorded_fps:
                die("source timing differs from plan evidence")
        if source.get("transport") not in {"https", "ephemeral-upload"}:
            die("source transport must be approved HTTPS or ephemeral-upload")
        if source.get("uriSha256") != sha256_bytes(uri.encode("utf-8")):
            die("payload URI does not match its source evidence")
        if source.get("transport") == "ephemeral-upload":
            parsed = urllib.parse.urlsplit(uri)
            if parsed.scheme != "runway" or parsed.query or parsed.fragment or not source.get("uploadReceiptSha256"):
                die("ephemeral upload URI/receipt evidence is invalid")
        else:
            approved_host = source.get("approvedHost")
            if not isinstance(approved_host, str):
                die("HTTPS source lacks its exact approved hostname")
            validate_public_https(uri, {approved_host})
    try:
        seconds = Decimal(str(plan.get("billingSeconds")))
    except InvalidOperation:
        die("source billingSeconds must be decimal")
    if not seconds.is_finite() or seconds <= 0:
        die("source billingSeconds must be finite and positive")
    if model in {"gen4.5", "gen4_turbo"}:
        if len(sources) != 1 or sources[0].get("role") != "first-frame" or sources[0].get("kind") != "image":
            die("image-to-video requires one validated first-frame image")
        if payload.get("ratio") not in {"1280:720", "720:1280", "1104:832", "832:1104", "1584:672", "960:960"}:
            die("image-to-video ratio is invalid")
        if model == "gen4.5" and not isinstance(payload.get("promptText"), str):
            die("Gen-4.5 image-to-video requires promptText")
        requested = payload.get("duration")
        if not isinstance(requested, int) or not 2 <= requested <= 10 or seconds != requested:
            die("image-to-video billingSeconds must equal integer duration 2..10")
        credits = Decimal(12 if model == "gen4.5" else 5) * seconds
    elif model == "aleph2":
        if len(payload.get("keyframes", [])) > 5:
            die("Aleph accepts no more than five keyframes")
        if sources[0].get("role") != "edit-video" or sources[0].get("kind") != "video" or seconds != Decimal(str(sources[0].get("durationSeconds"))):
            die("Aleph billing must equal the re-probed edit-video duration")
        if seconds < 2 or seconds > 30:
            die("Aleph verified source duration must be 2..30 seconds")
        if Decimal(str(sources[0].get("fps"))) > 30 or max(sources[0]["width"], sources[0]["height"]) > 1920 or min(sources[0]["width"], sources[0]["height"]) > 1080:
            die("Aleph source exceeds fps or 1080p constraints")
        credits = max(Decimal(56), Decimal(28) * seconds.to_integral_value(rounding=ROUND_CEILING))
    else:
        character = payload.get("character", {})
        reference = payload.get("reference", {})
        if (character.get("type") not in {"image", "video"} or reference.get("type") != "video"
                or payload.get("expressionIntensity", 3) not in {1, 2, 3, 4, 5}):
            die("Act-Two character/reference/expression fields are invalid")
        performances = [item for item in sources if item.get("role") == "consented-driving-performance" and item.get("kind") == "video"]
        if len(performances) != 1 or seconds != Decimal(str(performances[0].get("durationSeconds"))):
            die("Act-Two billing must equal the re-probed performance duration")
        if seconds < 3 or seconds > 30:
            die("Act-Two verified performance duration must be 3..30 seconds")
        credits = Decimal(5) * max(Decimal(3), seconds.to_integral_value(rounding=ROUND_CEILING))
    return model, endpoint, payload, sources, credits


def poll(task_id: str, deadline_seconds: int) -> dict:
    deadline = time.monotonic() + deadline_seconds
    while True:
        try:
            task = api_json("GET", f"/v1/tasks/{task_id}", None, retry_reads=True)
        except APIProblem as exc:
            die(f"poll failed: {exc.kind}; status={exc.status}; bodySha256={exc.body_sha256}")
        if task.get("id") != task_id:
            die("task response id mismatch")
        status = task.get("status")
        if status in TERMINAL:
            return task
        if status not in NONTERMINAL:
            die(f"unknown task status {status!r}; stop rather than guess")
        if time.monotonic() >= deadline:
            die(f"poll deadline reached for {task_id}; task remains live; resume later")
        time.sleep(5.0 + random.random() * 1.5)


def verify_mp4(path: Path) -> dict:
    size = path.stat().st_size
    if size <= 0 or size > MAX_VIDEO:
        die("artifact byte count is outside policy")
    with path.open("rb") as stream:
        if stream.read(12)[4:8] != b"ftyp":
            die("output lacks ISO BMFF ftyp signature")
        digest = hashlib.sha256()
        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
    )
    try:
        metadata = json.loads(raw)
    except json.JSONDecodeError:
        die("ffprobe returned invalid JSON")
    streams = metadata.get("streams")
    if not isinstance(streams, list) or not any(item.get("codec_type") == "video" for item in streams):
        die("output has no video stream")
    bounded_process(["ffmpeg", "-nostdin", "-v", "error", "-i", str(path), "-f", "null", "-"], 300)
    video = next(item for item in streams if item.get("codec_type") == "video")
    return {
        "bytes": size,
        "sha256": digest.hexdigest(),
        "qa": {
            "container": metadata.get("format", {}).get("format_name"),
            "videoCodec": video.get("codec_name"),
            "width": video.get("width"), "height": video.get("height"),
            "fps": video.get("avg_frame_rate"),
            "duration": metadata.get("format", {}).get("duration"),
            "audioPresent": any(item.get("codec_type") == "audio" for item in streams),
            "fullDecode": "passed",
        },
    }


def download_mp4(url: str, destination: Path, allowed_hosts: set[str], ledger: Path, record: dict) -> dict:
    destination.parent.mkdir(parents=True, exist_ok=True)
    stage = destination.with_name("." + destination.name + ".runway-stage")
    url_hash = sha256_bytes(url.encode("utf-8"))
    expected = record.get("artifactStaged") or record.get("artifact")

    if destination.exists():
        if not isinstance(expected, dict) or not expected.get("sha256"):
            die("existing destination has no durable expected digest; refusing adoption")
        artifact = verify_mp4(destination)
        if artifact["sha256"] != expected["sha256"] or artifact["bytes"] != expected["bytes"]:
            die("existing destination differs from the durable staged artifact")
        return artifact

    if stage.exists() and isinstance(expected, dict) and expected.get("sha256"):
        artifact = verify_mp4(stage)
        if artifact["sha256"] != expected["sha256"] or artifact["bytes"] != expected["bytes"]:
            die("staged artifact differs from the durable expected digest")
        os.replace(stage, destination)
        fsync_directory(destination.parent)
        return artifact

    host, target, addresses = validate_public_https(url, allowed_hosts)
    if record.get("outputUrlSha256") not in (None, url_hash):
        die("task output URL identity changed during artifact recovery")
    record.update({
        "state": "downloading", "outputUrlSha256": url_hash,
        "stageName": stage.name, "downloadStartedAt": record.get("downloadStartedAt") or utc_now(),
    })
    atomic_json(ledger, record)

    artifact = None
    if stage.exists():
        try:
            artifact = verify_mp4(stage)
        except SystemExit:
            stage.unlink()
    if artifact is None:
        connection = PinnedHTTPSConnection(host, addresses[0], timeout=60)
        try:
            connection.request("GET", target, headers={"Accept": "video/mp4", "Host": host})
            response = connection.getresponse()
            if response.status != 200:
                die(f"output returned HTTP {response.status}")
            media_type = response.getheader("Content-Type", "").split(";", 1)[0].strip().lower()
            if media_type != "video/mp4":
                die(f"reference client only accepts video/mp4, got {media_type!r}")
            announced = response.getheader("Content-Length")
            try:
                announced_size = int(announced) if announced is not None else None
            except ValueError:
                die("invalid output Content-Length")
            if announced_size is not None and (announced_size < 0 or announced_size > MAX_VIDEO):
                die("output exceeds byte cap")
            fd = os.open(stage, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
            total = 0
            try:
                with os.fdopen(fd, "wb") as stream:
                    fd = None
                    while True:
                        chunk = response.read(1024 * 1024)
                        if not chunk:
                            break
                        total += len(chunk)
                        if total > MAX_VIDEO:
                            die("output crossed byte cap")
                        stream.write(chunk)
                    stream.flush()
                    os.fsync(stream.fileno())
            except BaseException:
                if stage.exists():
                    stage.unlink()
                raise
            finally:
                if fd is not None:
                    os.close(fd)
        finally:
            connection.close()
        artifact = verify_mp4(stage)

    record.update({"state": "artifact_staged", "artifactStaged": artifact, "artifactValidatedAt": utc_now()})
    atomic_json(ledger, record)
    os.replace(stage, destination)
    fsync_directory(destination.parent)
    return artifact


def main() -> None:
    prompt = os.environ.get("RUNWAY_PROMPT", "").strip()
    ratio = os.environ.get("RUNWAY_RATIO", "1280:720")
    duration_text = os.environ.get("RUNWAY_DURATION", "")
    job_key = os.environ.get("RUNWAY_JOB_KEY", "").strip()
    output = Path(os.environ.get("RUNWAY_OUTPUT", "runway-output.mp4")).resolve()
    if not job_key or not all(character.isalnum() or character in "-_" for character in job_key):
        die("RUNWAY_JOB_KEY must contain only letters, digits, hyphen, underscore")

    output_hosts = exact_hosts(os.environ.get("RUNWAY_OUTPUT_HOSTS", ""), "RUNWAY_OUTPUT_HOSTS")
    rights_sha = evidence_sha("RUNWAY_RIGHTS_RECORD_SHA256")
    moderation_sha = evidence_sha("RUNWAY_MODERATION_RECORD_SHA256")
    qa_plan_sha = evidence_sha("RUNWAY_QA_PLAN_SHA256")
    governance_sha = evidence_sha("RUNWAY_GOVERNANCE_RECORD_SHA256")
    test_record_sha = evidence_sha("RUNWAY_TEST_RECORD_SHA256")

    source_plan_name = os.environ.get("RUNWAY_SOURCE_PLAN", "").strip()
    ledger_root = Path(os.environ.get("RUNWAY_LEDGER_DIR", ".runway-video-jobs")).resolve()
    if source_plan_name:
        model, endpoint, payload, sources, credits = load_source_plan(Path(source_plan_name).resolve())
        prompt = str(payload.get("promptText", ""))
    else:
        if not prompt or len(prompt.encode("utf-16-le")) // 2 > 1000:
            die("RUNWAY_PROMPT must be 1..1000 UTF-16 code units")
        if ratio not in {"1280:720", "720:1280"}:
            die("unsupported Gen-4.5 text-to-video ratio")
        try:
            duration = int(duration_text)
        except ValueError:
            die("RUNWAY_DURATION must be an integer")
        if not 2 <= duration <= 10:
            die("RUNWAY_DURATION must be 2..10")
        model, endpoint, sources = "gen4.5", CREATE_PATH, []
        payload = {"model": model, "promptText": prompt, "ratio": ratio, "duration": duration}
        credits = Decimal(PRICE_CREDITS_PER_SECOND) * Decimal(duration)
    usd = USD_PER_CREDIT * credits
    if not credits.is_finite() or credits <= 0 or not usd.is_finite() or usd <= 0:
        die("computed credits and USD must be finite and positive")
    output_policy = {
        "format": "video/mp4", "maxBytes": MAX_VIDEO, "fullDecode": True,
        "exactHosts": list(output_hosts), "redirects": "reject", "publicDns": True,
        "rejectIpLiterals": True, "dnsPinning": "connect-IP-with-TLS-SNI-and-host-certificate",
    }
    envelope = {
        "apiOrigin": API_ORIGIN,
        "apiVersion": API_VERSION,
        "downloadPolicy": output_policy,
        "endpoint": endpoint,
        "jobKey": job_key,
        "ledgerDir": str(ledger_root),
        "output": str(output),
        "payload": payload,
        "price": {
            "asOf": "2026-07-10",
            "credits": format(credits, "f"),
            "usdBeforeTax": format(usd, ".2f"),
        },
        "sources": sources,
        "evidence": {
            "rightsSha256": rights_sha, "moderationSha256": moderation_sha,
            "qaPlanSha256": qa_plan_sha, "governanceSha256": governance_sha,
            "testRecordSha256": test_record_sha,
        },
    }
    approval = sha256_bytes(canonical(envelope))
    safe_plan = dict(envelope)
    safe_plan["payload"] = redacted_payload(payload)
    safe_plan["sources"] = sanitized_sources(sources)
    print(json.dumps({"approvalSha256": approval, "plan": safe_plan}, indent=2))
    if os.environ.get("RUNWAY_SHOW_FULL_PLAN") == "YES":
        print(json.dumps({"protectedFullEnvelope": envelope}, indent=2))

    if os.environ.get("SEND_RUNWAY_REQUEST") != "1":
        print("DRY RUN: no network request sent", file=sys.stderr)
        return
    if os.environ.get("APPROVE_RUNWAY_PAID_CALL") != "YES":
        die("APPROVE_RUNWAY_PAID_CALL=YES is required")
    if os.environ.get("RUNWAY_APPROVAL_SHA256", "") != approval:
        die("approval digest mismatch")
    maximum = finite_positive_decimal("RUNWAY_MAX_COST_USD")
    if maximum < usd:
        die(f"cost bound {maximum} is below {usd}")

    ledger = ledger_root / f"{job_key}.json"
    request_hash = sha256_bytes(canonical({"endpoint": endpoint, "payload": payload}))
    if ledger.exists():
        record = read_json(ledger)
        if not set(record).issubset(LEDGER_KEYS):
            die("ledger contains a non-allowlisted field")
        if record.get("requestHash") != request_hash:
            die("job key already belongs to a different request")
        if record.get("approvalSha256") != approval:
            die("ledger approval differs from the current exact plan")
        task_id = record.get("taskId")
        if not task_id:
            die(f"{record.get('state', 'ambiguous_create')} ledger has no task id; do not recreate")
        task_id = validate_task_id(task_id)
    else:
        record = {
            "schemaVersion": 1, "state": "creating", "createdAt": utc_now(),
            "requestHash": request_hash, "approvalSha256": approval,
            "model": model, "endpoint": endpoint, "apiVersion": API_VERSION,
            "promptSha256": sha256_bytes(prompt.encode("utf-8")), "sources": sanitized_sources(sources),
            "outputPolicy": output_policy,
            "evidence": envelope["evidence"], "price": envelope["price"],
            "unresolvedGovernance": UNRESOLVED_GOVERNANCE,
        }
        try:
            exclusive_ledger(ledger, record)
        except FileExistsError:
            die("another process acquired this job key")
        try:
            created = api_json("POST", endpoint, payload, retry_reads=False)
        except APIProblem as exc:
            ambiguous = exc.status is None or exc.status >= 500 or exc.status in {408, 409, 429}
            record.update({
                "state": "ambiguous_create" if ambiguous else "create_rejected",
                "createOutcomeAt": utc_now(), "errorKind": exc.kind,
                "httpStatus": exc.status, "errorBodySha256": exc.body_sha256,
                "replayAllowed": False,
            })
            atomic_json(ledger, record)
            die(f"create stopped: {record['state']}; status={exc.status}; bodySha256={exc.body_sha256}")
        try:
            task_id = validate_task_id(created.get("id"))
        except SystemExit:
            record.update({
                "state": "ambiguous_create", "createOutcomeAt": utc_now(),
                "errorKind": "unusable-success", "responseSha256": sha256_bytes(canonical(created)),
                "replayAllowed": False,
            })
            atomic_json(ledger, record)
            die("ambiguous_create: unusable success response; do not recreate")
        record.update({"state": "submitted", "taskId": task_id, "submittedAt": utc_now()})
        atomic_json(ledger, record)

    task = poll(task_id, int(os.environ.get("RUNWAY_POLL_DEADLINE", "900")))
    record.update({"state": task["status"].lower(), "taskId": task_id, "terminalAt": utc_now()})
    if task["status"] == "FAILED":
        record["failureCode"] = task.get("failureCode")
        atomic_json(ledger, record)
        die(f"task failed: {task.get('failureCode')!r}")
    if task["status"] == "CANCELLED":
        atomic_json(ledger, record)
        die("task was cancelled")
    outputs = task.get("output")
    if not isinstance(outputs, list) or len(outputs) != 1 or not isinstance(outputs[0], str):
        record.update({"state": "artifact_blocked", "artifactError": "invalid-output-shape"})
        atomic_json(ledger, record)
        die("reference client requires exactly one output URL")
    artifact = download_mp4(outputs[0], output, set(output_hosts), ledger, record)
    actual_raw = os.environ.get("RUNWAY_ACTUAL_CREDITS", "").strip()
    if actual_raw:
        try:
            actual = Decimal(actual_raw)
        except InvalidOperation:
            die("RUNWAY_ACTUAL_CREDITS must be a decimal")
        if not actual.is_finite() or actual < 0:
            die("RUNWAY_ACTUAL_CREDITS must be finite and non-negative")
        reconciliation = {
            "status": "matched" if actual == credits else "variance-stop-further-creates",
            "estimatedCredits": format(credits, "f"), "actualCredits": format(actual, "f"),
            "varianceCredits": format(actual - credits, "f"),
        }
    else:
        reconciliation = {"status": "pending-stop-further-creates", "estimatedCredits": format(credits, "f")}
    manifest = {
        "schemaVersion": 1, "taskId": task_id, "jobKey": job_key, "state": "stored",
        "timestamps": {
            "created": record.get("createdAt"), "submitted": record.get("submittedAt"),
            "terminal": record.get("terminalAt"), "validated": record.get("artifactValidatedAt"),
            "stored": utc_now(),
        },
        "model": model, "endpoint": endpoint, "apiVersion": API_VERSION,
        "requestHash": request_hash, "approvalSha256": approval,
        "promptSha256": sha256_bytes(prompt.encode("utf-8")), "sources": sanitized_sources(sources),
        "outputPolicy": output_policy, "artifact": dict(artifact, path=str(output)),
        "price": envelope["price"], "costReconciliation": reconciliation,
        "rightsModeration": {"rightsSha256": rights_sha, "moderationSha256": moderation_sha},
        "qa": {"planSha256": qa_plan_sha, "automated": artifact["qa"], "humanReview": "pending"},
        "tests": {
            "recordSha256": test_record_sha, "ffprobe": "passed", "fullDecode": "passed",
            "adversarialSuite": ["non-finite-cost", "host-policy-binding", "DNS-IP-literal", "publish-crash-resume", "sensitive-error-redaction", "ambiguous-create", "tool-output-flood", "concurrency-status"],
        },
        "governance": {"recordSha256": governance_sha, "unresolved": UNRESOLVED_GOVERNANCE},
    }
    manifest_path = output.with_suffix(output.suffix + ".manifest.json")
    atomic_json(manifest_path, manifest)
    record.update({
        "state": "stored", "storedAt": manifest["timestamps"]["stored"],
        "artifact": artifact, "manifestSha256": sha256_bytes(canonical(manifest)),
        "costReconciliation": reconciliation,
    })
    atomic_json(ledger, record)
    print(json.dumps({"taskId": task_id, "artifact": dict(artifact, path=str(output)), "manifest": str(manifest_path)}, indent=2))


if __name__ == "__main__":
    main()

PowerShell でのドライランの例：

$env:RUNWAY_PROMPT = 'A matte cobalt travel mug stands on pale limestone in soft morning window light. A bead of condensation slowly rolls down its side. The camera performs a restrained 20-degree dolly arc in a single continuous medium close-up.'
$env:RUNWAY_RATIO = '1280:720'
$env:RUNWAY_DURATION = '6'
$env:RUNWAY_JOB_KEY = 'mug-reveal-v1'
$env:RUNWAY_OUTPUT = 'C:\approved-output\mug-reveal-v1.mp4'
$env:RUNWAY_OUTPUT_HOSTS = '<reviewed exact output hostname>'
$env:RUNWAY_RIGHTS_RECORD_SHA256 = '<64-hex protected rights-record digest>'
$env:RUNWAY_MODERATION_RECORD_SHA256 = '<64-hex protected moderation-record digest>'
$env:RUNWAY_QA_PLAN_SHA256 = '<64-hex protected QA-plan digest>'
$env:RUNWAY_GOVERNANCE_RECORD_SHA256 = '<64-hex protected governance-record digest>'
$env:RUNWAY_TEST_RECORD_SHA256 = '<64-hex protected adversarial-test-record digest>'
python .\runway_video.py

予想コスト：72 クレジット、税別 $0.72 です。保護された完全なエンベロープを確認してから、表示されたダイジェストをコピーし、意図的に有料呼び出しを実行してください。

$env:SEND_RUNWAY_REQUEST = '1'
$env:APPROVE_RUNWAY_PAID_CALL = 'YES'
$env:RUNWAY_APPROVAL_SHA256 = '<exact digest from this unchanged plan>'
$env:RUNWAY_MAX_COST_USD = '0.72'
$env:RUNWAYML_API_SECRET = '<secret injected by a secret manager>'
python .\runway_video.py

実際のシークレットをシェルの履歴に残さないでください。本番環境では、プロセススーパーバイザーまたはシークレットマネージャーから注入してください。

ネイティブのペイロードテンプレート

以下は、調査時点での完全なリクエスト本文です。有料リクエストの送信を許可するものではありません。同じプラン、ソースハッシュ、予算、承認、台帳の管理手順を適用してください。

Gen-4.5、最初のフレームを使った画像から動画への変換、5秒、税抜0.60ドル：

{
  "model": "gen4.5",
  "promptImage": [{"uri": "runway://APPROVED_UPLOAD", "position": "first"}],
  "promptText": "The subject lifts the mug once and takes a small sip. The locked camera holds a medium close-up. Steam drifts gently upward in soft morning light.",
  "ratio": "1104:832",
  "duration": 5,
  "seed": 18421,
  "contentModeration": {"publicFigureThreshold": "auto"}
}

Gen-4 Turbo、画像から動画への変換、6秒、税抜0.30ドル：

{
  "model": "gen4_turbo",
  "promptImage": "runway://APPROVED_UPLOAD",
  "promptText": "The fabric ripples in a light breeze while the camera slowly pushes in.",
  "ratio": "1280:720",
  "duration": 6,
  "contentModeration": {"publicFigureThreshold": "auto"}
}

Aleph 2.0、絶対キーフレームを1つ使用し、検証済み入力の全長を基に保守的に料金を算出：

{
  "model": "aleph2",
  "videoUri": "runway://APPROVED_VIDEO_UPLOAD",
  "promptText": "Relight the room with cool moonlight while preserving the performer, wardrobe, camera motion, and set layout.",
  "keyframes": [
    {"uri": "runway://APPROVED_KEYFRAME_UPLOAD", "seconds": 2.5}
  ],
  "seed": 18421,
  "contentModeration": {"publicFigureThreshold": "auto"}
}

Act-Two、キャラクター画像と8秒の演技を使用、税抜0.40ドルが上限：

{
  "model": "act_two",
  "character": {"type": "image", "uri": "runway://CONSENTED_CHARACTER_IMAGE"},
  "reference": {"type": "video", "uri": "runway://CONSENTED_DRIVING_PERFORMANCE"},
  "bodyControl": true,
  "expressionIntensity": 3,
  "ratio": "720:1280",
  "seed": 18421,
  "contentModeration": {"publicFigureThreshold": "auto"}
}

画像から動画への変換、Aleph、Act-Two の実行可能なソースパイプライン

本番リファレンスを runway_video.py として保存し、次のコンパニオンを runway_source_plan.py として保存してください。このコンパニオンは、ローカルのミラーファイルを検証して完全にデコードし、DNSピン留めとリダイレクトなしの HEAD によりレビュー済みのHTTPSソースを確認するか、すでに完了した公式の一時アップロードについてレシートダイジェストを要求します。ソースプランを書き出します。RUNWAY_SOURCE_PLAN を設定すると、本番クライアントはすべてのローカルミラーのハッシュを再計算し、保守的な価格を再計算して、正確なソースペイロードと証拠を承認に紐付けてから、同じ排他的作成、ポーリング、クラッシュから再開可能なアーティファクトのライフサイクルを実行します。アップロードは、意図的に別途記録されるステートフルな操作です。マルチパートアップロードに失敗した場合は新しいセッションを開始し、盲目的に再試行することはありません。

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

import json
import os
import urllib.parse
from decimal import Decimal
from pathlib import Path

import runway_video as rv

IMAGE_RATIOS = {"1280:720", "720:1280", "1104:832", "832:1104", "1584:672", "960:960"}


def required(name: str) -> str:
    value = os.environ.get(name, "").strip()
    if not value:
        rv.die(f"{name} is required")
    return value


def probe_local(path: Path, kind: str) -> dict:
    if not path.is_file():
        rv.die(f"missing local {kind} mirror")
    with path.open("rb") as stream:
        header = stream.read(32)
    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")
    ):
        rv.die("image must be JPEG, PNG, or WebP by signature")
    raw = rv.bounded_process(
        ["ffprobe", "-v", "error", "-show_streams", "-show_format", "-of", "json", str(path)], 30, rv.MAX_JSON
    )
    try:
        metadata = json.loads(raw)
    except json.JSONDecodeError:
        rv.die("ffprobe returned invalid JSON")
    streams = metadata.get("streams", [])
    video = next((item for item in streams if item.get("codec_type") == "video"), None)
    if not video:
        rv.die(f"{kind} mirror has no visual stream")
    rv.bounded_process(["ffmpeg", "-nostdin", "-v", "error", "-i", str(path), "-f", "null", "-"], 300)
    result = {"width": video.get("width"), "height": video.get("height")}
    if not all(isinstance(result[key], int) and result[key] > 0 for key in ("width", "height")):
        rv.die("invalid source dimensions")
    if kind == "video":
        try:
            duration = Decimal(str(metadata.get("format", {}).get("duration")))
            numerator, denominator = str(video.get("avg_frame_rate", "0/1")).split("/", 1)
            fps = Decimal(numerator) / Decimal(denominator)
        except Exception:
            rv.die("video duration/fps are unusable")
        if not duration.is_finite() or duration <= 0 or not fps.is_finite() or fps <= 0:
            rv.die("video duration/fps must be finite and positive")
        result.update({"durationSeconds": format(duration, "f"), "fps": format(fps, "f")})
    return result


def checked_source(index: int, role: str, kind: str, input_hosts: set[str]) -> tuple[str, dict]:
    prefix = f"RUNWAY_SOURCE_{index}_"
    local = Path(required(prefix + "LOCAL")).resolve()
    uri = required(prefix + "URI")
    details = probe_local(local, kind)
    size = local.stat().st_size
    parsed = urllib.parse.urlsplit(uri)
    if parsed.scheme == "runway":
        if not uri.startswith("runway://") or len(uri) > 2048 or parsed.query or parsed.fragment:
            rv.die("invalid ephemeral-upload URI")
        receipt = required(prefix + "UPLOAD_RECEIPT_SHA256").lower()
        if len(receipt) != 64 or any(character not in "0123456789abcdef" for character in receipt):
            rv.die("upload receipt must be a SHA-256")
        transport, host = "ephemeral-upload", None
        cap = 200 * 1024 * 1024
    elif parsed.scheme == "https":
        host, target, addresses = rv.validate_public_https(uri, input_hosts)
        connection = rv.PinnedHTTPSConnection(host, addresses[0], timeout=10)
        try:
            connection.request("HEAD", target, headers={"Host": host})
            response = connection.getresponse()
            if response.status != 200:
                rv.die(f"source HEAD returned HTTP {response.status}; redirects are rejected")
            announced = response.getheader("Content-Length")
            media_type = response.getheader("Content-Type", "").split(";", 1)[0].lower()
            expected_types = {"image": {"image/jpeg", "image/png", "image/webp"}, "video": {"video/mp4", "video/quicktime", "video/webm"}}
            if announced is None or int(announced) != size or media_type not in expected_types[kind]:
                rv.die("source HEAD length/type does not match the validated local mirror")
        finally:
            connection.close()
        receipt, transport = None, "https"
        cap = (16 if kind == "image" else 32) * 1024 * 1024
    else:
        rv.die("source URI must be reviewed HTTPS or runway:// ephemeral upload")
    if size < 512 or size > cap:
        rv.die("source byte count is outside the selected transport limit")
    evidence = {
        "role": role, "kind": kind, "transport": transport, "localPath": str(local),
        "bytes": size, "sha256": rv.file_sha256(local), "uriSha256": rv.sha256_bytes(uri.encode()),
        "contentType": "locally-probed", "validatedAt": rv.utc_now(), **details,
    }
    if host:
        evidence["approvedHost"] = host
    if receipt:
        evidence.update({"uploadReceiptSha256": receipt, "uploadState": "completed-do-not-retry-session"})
    return uri, evidence


def main() -> None:
    mode = required("RUNWAY_SOURCE_MODE")
    input_hosts = set(rv.exact_hosts(os.environ.get("RUNWAY_INPUT_HOSTS", ""), "RUNWAY_INPUT_HOSTS"))
    prompt = os.environ.get("RUNWAY_PROMPT", "").strip()
    ratio = os.environ.get("RUNWAY_RATIO", "1280:720")
    if ratio not in IMAGE_RATIOS:
        rv.die("unsupported ratio")
    if len(prompt.encode("utf-16-le")) // 2 > 1000:
        rv.die("prompt exceeds 1,000 UTF-16 code units")
    sources = []
    if mode in {"gen45-image", "gen4-turbo"}:
        uri, source = checked_source(1, "first-frame", "image", input_hosts)
        sources.append(source)
        duration = int(required("RUNWAY_DURATION"))
        if not 2 <= duration <= 10:
            rv.die("duration must be 2..10")
        model = "gen4.5" if mode == "gen45-image" else "gen4_turbo"
        if model == "gen4.5" and not prompt:
            rv.die("Gen-4.5 image-to-video requires RUNWAY_PROMPT")
        endpoint = "/v1/image_to_video"
        payload = {"model": model, "promptImage": uri, "ratio": ratio, "duration": duration, "contentModeration": {"publicFigureThreshold": "auto"}}
        if prompt:
            payload["promptText"] = prompt
        billing = Decimal(duration)
    elif mode == "aleph2":
        uri, source = checked_source(1, "edit-video", "video", input_hosts)
        sources.append(source)
        billing = Decimal(source["durationSeconds"])
        if (billing < 2 or billing > 30 or Decimal(source["fps"]) > 30
                or max(source["width"], source["height"]) > 1920
                or min(source["width"], source["height"]) > 1080):
            rv.die("Aleph source must be 2..30 s, <=30 fps, and <=1080p")
        payload = {"model": "aleph2", "videoUri": uri, "contentModeration": {"publicFigureThreshold": "auto"}}
        if prompt:
            payload["promptText"] = prompt
        if os.environ.get("RUNWAY_SOURCE_2_LOCAL"):
            key_uri, key_source = checked_source(2, "absolute-keyframe", "image", input_hosts)
            sources.append(key_source)
            at = finite = Decimal(required("RUNWAY_KEYFRAME_SECONDS"))
            if not finite.is_finite() or finite < 0 or finite > billing:
                rv.die("keyframe seconds must be finite and inside the source")
            payload["keyframes"] = [{"uri": key_uri, "seconds": float(at)}]
        model, endpoint = "aleph2", "/v1/video_to_video"
    elif mode == "act-two":
        character_kind = required("RUNWAY_CHARACTER_TYPE")
        if character_kind not in {"image", "video"}:
            rv.die("RUNWAY_CHARACTER_TYPE must be image or video")
        character_uri, character = checked_source(1, "consented-character", character_kind, input_hosts)
        performance_uri, performance = checked_source(2, "consented-driving-performance", "video", input_hosts)
        sources.extend([character, performance])
        billing = Decimal(performance["durationSeconds"])
        if billing < 3 or billing > 30:
            rv.die("Act-Two performance must be 3..30 seconds")
        intensity = int(os.environ.get("RUNWAY_EXPRESSION_INTENSITY", "3"))
        if not 1 <= intensity <= 5:
            rv.die("expression intensity must be 1..5")
        payload = {
            "model": "act_two", "character": {"type": character_kind, "uri": character_uri},
            "reference": {"type": "video", "uri": performance_uri}, "bodyControl": character_kind == "image",
            "expressionIntensity": intensity, "ratio": ratio, "contentModeration": {"publicFigureThreshold": "auto"},
        }
        model, endpoint = "act_two", "/v1/character_performance"
    else:
        rv.die("RUNWAY_SOURCE_MODE must be gen45-image, gen4-turbo, aleph2, or act-two")
    plan = {"schemaVersion": 1, "model": model, "endpoint": endpoint, "payload": payload, "sources": sources, "billingSeconds": format(billing, "f")}
    rv.atomic_json(Path(required("RUNWAY_SOURCE_PLAN_OUT")).resolve(), plan)
    print(json.dumps({"sourcePlanSha256": rv.sha256_bytes(rv.canonical(plan)), "sourceCount": len(sources)}, indent=2))


if __name__ == "__main__":
    main()

ドライランの前に、生成されたファイルを RUNWAY_SOURCE_PLAN に設定してください。承認を計算する前に、正確な出力ホストポリシーと、保護対象となる4つすべての証拠ダイジェストも設定してください。ソースURI、署名付きクエリ、顔、プロンプト、アップロードハンドルを通常のログに記録しないでください。コンパニオンが出力するのはプランダイジェストと件数のみです。

publicFigureThreshold: "low" は、認識可能な著名人に対する制限が緩やかになると説明されています。これは同意要件の回避策ではなく、ポリシーや権利の確認を無効化するために決して使用してはいけません。デフォルトは auto にしてください。low を使用する場合は、文書化された許可済みの用途と、更新された承認が必要です。

障害への対処

人間が読める failure 文字列より先に failureCode を確認してください。

• SAFETY.* と INPUT_PREPROCESSING.SAFETY.TEXT — ポリシー／モデレーションによる失敗です。同じリクエストを再試行したり、難読化したりしないでください。独自のスクリーニングを見直してください。モデレーション対象のリクエストを繰り返すと、アカウントが停止される可能性があり、料金も発生します。
• ASSET.INVALID — ソースの利用可否、形式、サイズ、寸法、長さ、fps、アスペクト比を検証してください。新しい承認ダイジェストでアセットを修正してください。
• INPUT_PREPROCESSING.INTERNAL — 入力の処理に失敗しました。まずアセットを確認してください。再試行が妥当な場合は、新しい明示的な有料承認とジョブキーを使用してください。
• THIRD_PARTY.UNAVAILABLE — パートナーモデルに関係します。待機するか、ユーザーの許可と新しいダイジェストを得た場合に限り、モデルを選択してください。
• INTERNAL.BAD_OUTPUT、INTERNAL、または null — 一時的な問題の可能性がありますが、再試行は Runway が返金していない限り、別の有料作成になります。クレジットの使用状況を確認し、新たな承認を得てください。

Runway のウェブヘルプでは通常の生成エラーは返金されると説明されていますが、API のモデレーション文書では、モデレーション対象の生成には成功時と同じ料金がかかると明記されています。ウェブアプリの返金に関する説明を、すべてのAPI失敗コードに一般化しないでください。公開API文書には、完全な返金一覧がありません。再試行する前にクレジットを照合してください。

権利、音声、来歴、データガバナンス

権利と肖像

Runway の利用規約では、ユーザーの入力または出力の所有権を主張せず、規約への準拠を条件に出力の商用利用を制限しないと定めています。これはプロバイダーとユーザー間の責任分担であり、著作権の成立、独占性、商標のクリアランス、パブリシティ権、音楽の権利、または第三者の作品との類似性がないことを保証するものではありません。

すべてのソース、ライセンサー、ライセンスの範囲、同意、地域、期間、モデルリリース、音声リリース、および予定している配信を明記した権利台帳を維持してください。Act-Two では、描写されるキャラクターと、演技や音声を提供する人物の両方から許可を得る必要があります。Runway の利用ポリシーで禁止されている場合は、存命のアーティストの作風を使用しないでください。

音声

gen4.5 と gen4_turbo の現在の API スキーマには、音声生成の切り替え項目はありません。ネイティブ音声に言及する一般向け UI やマーケティング資料は、API の仕様を保証するものではありません。更新された API スキーマに音声フィールドが追加されない限り、ライセンスを取得したサウンドデザインを後処理で追加する計画にしてください。

Act-Two の製品ガイドには、演技を提供する人物のパフォーマンスから音声の特徴が転送されると記載されていますが、API スキーマには独立した音声切り替え項目はありません。返されたストリームを確認し、同意を得た素材でテストするまでは、リップシンク、音声の識別性、または音声の保持を保証しないでください。パートナーモデルには、それぞれ独自の音声フラグと価格が設定されています。

帰属表示と開示

API を統合したエンドユーザー向けインターフェースでは、利用規約と帰属表示ページに従い、「Powered by Runway」を目立つ形で表示し、runwayml.com にリンクする必要があります。Runway の一般ユーザー向け利用権 FAQ では、完成作品に正式なクレジットは不要とされていますが、これは API 統合 UI における表示義務に優先するものではありません。

Runway は C2PA の来歴シグナルを実装していると述べていますが、確認した公開 API の出力仕様では、ダウンロードしたすべてのファイルに C2PA が含まれるとは保証されていません。埋め込みメタデータを保持し、取得前に情報を損なうトランスコードを避け、必要に応じて C2PA 検証ツールで確認し、独自の署名付き来歴マニフェストを保管してください。文脈によって誤解を招くおそれがある場合は、現実的な合成の人物や出来事であることを明確に表示してください。

プライバシーと保持

Runway の公開 API マーケティングページには「お客様のデータをトレーニングに使用しない」とありますが、標準利用規約では、入力と出力をモデルのトレーニングおよび改善に使用する場合があると定めています。2026 年の Runway Dev の告知では、契約上のトレーニング不使用の約束とゼロデータ保持への対応について説明されており、エンタープライズ向けページでは別個の保護措置が説明されています。これらの公開情報だけでは、任意のセルフサービスアカウントに適用される条件は完全には明確になっていません。

トレーニングへの使用、保持、ゼロ保持、データ保管地域、削除 SLA、人によるレビュー、パートナーのサブプロセッサー、およびインシデント通知については、組織に適用される注文書、利用規約のバージョン、DPA によって解決されるまで、不明として扱ってください。マーケティング上の見出しを根拠に、機密情報、規制対象情報、生体情報、未成年者、またはクライアントによる利用制限のあるメディアを送信しないでください。

プライバシーポリシーでは、送信された顔と音声は生体データに該当する可能性があるとされ、顔スキャンと音声の特徴量は、目的の達成時または最後のやり取りから 3 年後のいずれか早い時点まで保持されると記載されています。また、顧客の指示による特定の処理は、公開プライバシーポリシーではなく顧客との契約に基づくとされています。必要な生体情報の通知と同意を取得し、該当する場合は DPIA を実施し、ソースを最小限に抑え、プロバイダー側とローカルシステムの両方で削除スケジュールを設定してください。

タスクを削除すると、キャンセルと同じ DELETE エンドポイントが使用され、タスクにアクセスできなくなります。すべてのログ、安全記録、バックアップ、または法的保存対象のコピーが消去されるとは文書化されていません。規制対象の消去には、契約上の削除手段を利用してください。

本番環境への導入チェックリスト

• ネイティブモデルとパートナーモデルのどちらを使用するかが明確です。
• 現在のモデル ID、エンドポイント、API バージョン、スキーマ、料金、提供終了状況を再確認しました。
• すべての入力について同意とライセンスを確認し、ローカルで検証し、ハッシュを取得し、不要な機微データが含まれていないことを確認しました。
• プロンプトと参照素材の方針がモードに適合しています。
• ドライランのコストと、保護された完全なペイロードを確認しました。
• 正確な承認ダイジェスト、最大コスト、排他的なジョブキーが設定されています。
• 作成リクエストの再試行は無効になっており、作成結果が不明な場合に自動で再作成されることはありません。
• タスク ID は永続的に保存され、ポーリングでは THROTTLED と CANCELLED が適切に処理されます。
• ポーリングのタイムアウトによってキャンセルされることはありません。DELETE には、個別の破壊的操作に対する承認が必要です。
• 有料 API 呼び出しの前にモデレーションを実施し、ブロックされたリクエストを回避していません。
• 出力は有効期限前にダウンロードし、ホスト、サイズ、種類を検証したうえで、完全にデコードし、ハッシュを取得して、永続的に保存しています。
• 音声ストリーム、長さ、fps、寸法、連続性、アーティファクト、受け入れ基準を人間が確認しています。
• 該当する API 連携 UI に「Runway を利用」と表示されています。
• 権利、同意、開示、C2PA の検査、来歴、データ利用条件、保持、削除の証拠を記録しています。
• クレジット使用量を見積もりと照合し、差異があれば以降の作成を停止します。

公式情報源

特に記載がない限り、2026-07-10 にアクセス：

• 利用可能な AI モデル
• API 変更履歴
• API 料金
• API の使用
• API リファレンス
• 公式 Python SDK と OpenAPI 生成型
• SDK のタスクポーリングに関するガイダンス
• API 入力
• 一時アップロード
• API 出力
• コンテンツモデレーション
• タスクの失敗
• HTTP エラー
• 利用ティア
• クレジット表記
• Gen-4.5 のプロンプトガイド
• テキストから動画を生成するプロンプトガイド
• Aleph 2.0 のプロンプトガイド
• Act-Two ガイド
• 参照メディアのガイドライン
• 利用規約
• 利用ポリシー
• プライバシーポリシー
• Runway のセキュリティ
• Runway Dev の発表
• エンタープライズ向けサードパーティモデル FAQ
• 利用権に関する FAQ
• 安全性と来歴