runway-video

Runway 视频

对于使用文档所述 Runway Developer API（https://api.dev.runwayml.com）的服务端集成，请使用此技能。它涵盖 Runway 原生视频模型，并明确将合作伙伴模型、静态图像生成、消费级 Web 应用、Recipes、Workflows、Characters 和第三方网关列为范围外内容。

研究截止日期：2026-07-10。

证据措辞

请按以下方式理解此技能中的陈述：

• 事实 — Runway 一手资料或官方 OpenAPI 生成的 SDK 中明确说明的内容。
• 声明 — Runway 的营销或政策措辞，可能因套餐或合同而异。
• 启发式建议 — 根据事实得出的操作建议；请针对部署进行验证。
• 未知 — 经审阅的 Runway 一手公开资料尚未证实的内容。请勿擅自补充。

当不稳定的事实会影响费用、安全、有效载荷有效性或数据治理时，生产使用前请重新查阅引用的一手页面。

从范围、同意和执行模式着手

设计请求前，请确认以下所有事项：

1. 期望的模式：文本转视频、图像转视频、视频编辑或表演迁移。
2. 是否必须使用 Runway 原生模型，还是可以接受合作伙伴模型。
3. 输出时长、尺寸、交付格式、音频要求和迭代预算。
4. 对所有提示词、图像、视频、人脸、声音、表演、标记和音乐输入的使用权。
5. 每位出镜或出声者的明确许可。Runway 的 Usage Policy 禁止未经他人许可使用其图像、视频或音频。
6. 当逼真的合成媒体可能误导观众时，制定披露/来源说明方案。
7. 数据分类、获批地区/子处理方、保留要求，以及组织合同是否确实包含不用于训练或零数据保留条款。
8. 以下执行模式之一：
   • 计划/试运行 — 默认模式。进行验证、定价并输出审批摘要。不创建任何付费 API 请求。
   • 获批创建 — 仅当精确请求摘要、限定的最高费用和明确的环境门控均匹配时执行。
   • 恢复 — 轮询已记录的任务 ID；绝不创建替代任务。
   • 取消/删除 — 单独的破坏性操作，需单独审批。绝不因超时而推断需要执行此操作。

拒绝缺少必要权利或许可、试图规避审核、冒充他人、欺诈、未经同意的私密影像、涉及未成年人的色情内容或其他禁止类别的请求。不要通过混淆内容来“修复”被拦截的内容。

路由请求

当前 Runway 原生视频功能

模型	文档所述 API 模式	端点	必需参数	当前公开价格
gen4.5	文本 → 视频	POST /v1/text_to_video	promptText、ratio、整数 duration，范围 2–10	12 积分/秒
gen4.5	首帧图像 + 文本 → 视频	POST /v1/image_to_video	promptImage、promptText、ratio、整数 duration，范围 2–10	12 积分/秒
gen4_turbo	首帧图像（+ 可选动作文本）→ 视频	POST /v1/image_to_video	promptImage、ratio；时长 2–10	5 积分/秒
aleph2	现有视频 + 可选文本/关键帧 → 编辑后的视频	POST /v1/video_to_video	videoUri；最多五个带时间的关键帧	28 积分/秒，最低 56 积分
act_two	角色图像/视频 + 驱动表演视频 → 视频	POST /v1/character_performance	character、reference，模型 act_two	5 积分/秒；公开的 Act-Two 指南规定最低时长为 3 秒

事实： 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" 的数组项。
• **Gen-4 Turbo：**必须提供图像输入。它支持上述六种比例，时长可以是 2 到 10 秒之间的任意值。这是成本较低的原生图像驱动选项。
• **Aleph 2.0：**输入视频时长必须为 2–30 秒，帧率不超过 30 fps，分辨率不高于 1080p。输出会保留输入分辨率，最高为 1080p。API 接受可选的 promptText、seed、contentModeration、targetAspectRatio 和最多五个关键帧。每个关键帧都使用 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 的企业版常见问题页面称，其企业条款和 DPA 适用于合作伙伴模型，且提供方承诺不使用数据训练；但这并不能证明每个自助 API 账户都适用相同条款。

明确不在范围内

• 静态图像路由（/v1/text_to_image）是暂存选项，不用于视频生成。静态图像调用需单独计费，并且需要自己的技能／工作流和审批。
• 消费者网页版应用、Explore Mode、Apps、Edit Studio UI 和网页版积分均不属于 API 合同范围。网页版应用积分与 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 文本转视频，同时描述画面中可见的内容及其运动方式。对于图像转视频，图像已经确定了主体、构图、色彩、光线和风格，因此应着重描述主体、摄像机和环境的运动。迭代时一次只添加一个变量。

经验法则：

• 短片段尽量采用一个镜头和一个主要动作。
• 使用具体的摄像机术语：“固定摄像机，” “缓慢推轨，” “手持跟拍，” 或 “俯拍静态画面。”
• 将否定表述改为正向效果。用 “固定摄像机” 代替 “摄像机不移动。”
• 不要要求模型可靠地渲染重要的法律文案或 Logo；应在后期合成经批准的文字和标识。
• 仅在比较提示词变体时固定种子。重复使用种子只能帮助保持相似性，不能保证结果完全确定。

Gen-4.5 文本提示词示例：

一只哑光钴蓝色旅行杯立在浅色石灰岩上，沐浴在柔和的清晨窗光中。一滴冷凝水沿杯身缓缓滑落。摄像机以克制的方式沿弧线推轨 20 度，整个过程保持为一个连续的中近景镜头。自然的商业电影摄影风格，产品轮廓清晰，背景柔和虚化。

对于 Aleph，使用动作动词加上有针对性的变换：“用清冷的月光重新照亮房间，同时保留演员、服装、摄像机运动和布景布局。” 从最小幅度的编辑开始。外观需要精确时使用关键帧；对于关键帧中不可见的运动或变换，使用 promptText。

对于 Act-Two，效果主要取决于参考素材，而不是文字描述：仅一个主体，始终能看到面部，腰部以上或更近景，光线均匀，没有剪切；如果启用了身体控制，开始时应露出双手；动作自然。将 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；不要接受最终用户提供的 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。分片上传失败后不得重试；应创建新的上传会话。上传具有状态，因此应记录 URI，但不要将 API 密钥或源数据写入日志。

输出媒体

成功的任务会返回包含一个或多个 URL 的 output 数组。Runway 表示，这些 URL 会在 24–48 小时内过期，必须下载到持久存储中；不要将它们作为产品 URL 暴露。官方 SDK 类型说明，重新获取任务时可能会返回新的 URL。

将输出 URL 视为不可信的签名 URL：

• 绝不要向其附加 Authorization。
• 要求使用 HTTPS、无用户信息、默认端口、公共 DNS、精确的输出主机允许列表；限制重定向次数并重新验证每一跳；限制字节数；检查预期媒体类型和容器签名。
• 下载到私有临时文件，计算 SHA-256，运行 ffprobe，使用 ffmpeg -v error -i … -f null - 完整解码，然后原子发布。
• 存储任务 ID、请求摘要、模型、端点、提示词哈希或受保护的提示词、源哈希、审批记录、时间戳、输出哈希、编解码器／尺寸／fps／时长／是否存在音频流，以及披露／权利记录。除非安全审查过的事件处理工作流有此需要，否则不要存储签名 URL 的查询字符串。

公共 API 输出页面并未承诺当前所有模型都使用同一种容器／编解码器。仅支持 MP4 的验证器遇到其他输出类型时必须采取失败关闭策略，而不是重命名文件。

创建前计算价格

适用税费前，开发者积分的公开价格为每个 $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 文档或合同另有确认，否则按完整源时长计算价格。

示例：

请求	积分	税前美元金额
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 并继续排队；超过每日创建限制时，创建请求会返回 429。文档没有说明通用的每分钟请求数限制，但这并不意味着可以在没有自有队列、预算和滥用控制的情况下创建任务。

精确的付费审批

付费请求在创建时必须满足以下所有条件：

1. 试运行生成了规范请求封装，其中包含端点、模型、精确的 JSON 负载、输入 SHA-256 值、本地作业键、输出策略、当前价格来源日期，以及保守的最高美元费用。
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 只发送一次。禁用创建调用的自动重试。
• 立即、持久地保存返回的任务 ID。
• 如果客户端超时、断开连接、崩溃、收到格式错误的成功响应，或在可能已发送请求字节但尚未存储任务 ID 时收到 5xx，则将任务标记为状态不明。不要自动再次创建。通过服务提供商支持渠道／使用记录进行核对，或在新的任务键下接受经人工授权的潜在重复任务。
• 如果已知任务 ID，则继续调用 GET /v1/tasks/{id}。绝不要仅因为轮询超时就用新生成的任务替代它。

官方 SDK 默认会对连接错误、408、409、429 和 5xx 重试两次。这对读取操作很方便，但对于没有服务提供商幂等性保障的付费创建操作并不安全。如果使用 SDK，请将创建客户端实例化为 max_retries=0；单独的读取客户端可以对安全的 GET 请求进行重试。

轮询间隔不得短于 5 秒。对可重试的读取错误使用抖动和指数退避。准确识别以下状态：

状态	含义	操作
PENDING	已接受，等待中	继续轮询
THROTTLED	已存储，但因并发限制尚未加入队列	继续轮询；不要重新创建
RUNNING	正在运行；可能包含进度	继续轮询
SUCCEEDED	终态；输出 URL 已可用	立即下载并验证
FAILED	终态；检查 failureCode 和经过清理的 failure	不要盲目重试
CANCELLED	终态	停止

官方 SDK 架构中的 CANCELLED 使用两个 “l” 字符。

轮询截止时间或 SDK 等待超时不会取消任务。DELETE /v1/tasks/{id} 的用途不止一种：它会取消 PENDING、THROTTLED 或 RUNNING 任务，但会删除其他任务及其持久化输出数据。如果任务在读取与删除两次调用之间完成，则读取后再删除会产生竞态。绝不要自动调用 DELETE；必须另行明确批准破坏性操作，并确认可能删除终态任务。公开文档没有说明取消时的退款／收费规则，因此取消不能保证控制成本。

完整生产参考实现

以下标准库 Python 客户端直接实现 Gen-4.5 文本转视频，并接受经过验证的源计划，以支持原生图像转视频、Aleph 或 Act-Two。它默认以试运行模式运行，将确切审批绑定到付费负载、源证据、价格、输出主机／网络策略和治理证据，使用独占式本地台账，不重试创建操作，安全地重试轮询，不会自动取消任务，不携带 API Bearer Token 下载文件，并通过 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 个 credits，税前 $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

不要将实际密钥写入 Shell 历史记录。在生产环境中，请通过进程监管程序或密钥管理器注入密钥。

原生负载模板

以下是截至研究截止日期的完整请求正文。它们并不授权发送付费请求；请使用相同的计划、源哈希、预算、审批和账本控制来执行这些请求。

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，使用一个绝对关键帧，按经过验证的整个输入时长保守计价：

{
  "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 设为生成的文件。计算审批前，还要设置精确的输出主机策略和全部四个受保护的证据摘要。切勿将源 URI、签名查询、面孔、提示词或上传句柄写入普通日志；配套文件只会打印计划摘要和数量。

文档说明，publicFigureThreshold: "low" 对可识别公众人物的限制较少。这并不代表可以绕过同意要求，绝不可用于规避政策或权利检查。默认使用 auto；使用 low 时，必须有记录在案的授权用途并重新审批。

故障处理

优先使用 failureCode，而不是面向用户的 failure 字符串：

• 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 面向消费者的使用权常见问题解答称，完成的作品无需正式署名；但这不免除 API 集成界面的义务。

Runway 称其采用 C2PA 来源信号，但经审查的公开 API 输出合约并未保证每个下载文件都包含 C2PA。保留任何嵌入的元数据，在捕获前避免破坏性转码；需要时使用 C2PA 验证工具进行检查，并保存自己的签名来源清单。在情境可能造成误导时，明确标注逼真的合成人物／事件。

隐私与保留

Runway 的公开 API 营销页面宣称“不会使用你的数据进行训练”，而标准条款则规定，输入和输出可能会用于训练和改进模型。Runway 于 2026 年发布的开发者公告提及合约中的不训练承诺和零数据保留支持，企业页面也介绍了单独的保护措施。对于任意自助服务账户，这些公开声明尚未完全协调一致。

在适用于组织的订单表单、条款版本和 DPA 明确规定之前，将训练用途、数据保留、零保留、数据驻留、删除 SLA、人工审核、合作方分处理商及事件通知视为未知。不要仅凭营销标题，就发送机密、受监管、生物识别、未成年人或受客户限制的媒体。

隐私政策将提交的人脸和语音视为可能的生物识别数据，并称人脸扫描／声纹的保留期限不超过目的完成时间或最后一次互动后三年的较早者；政策还称，某些由客户指示进行的处理受客户协议而非公开隐私政策约束。取得所需的生物识别通知／同意；在适用时完成 DPIA；尽量减少素材来源，并在提供方和本地系统中设置删除时间表。

删除任务使用与取消任务相同的 DELETE 端点，并会使任务无法访问。文档并未说明此操作会删除所有日志、安全记录、备份或依法保留的副本。对于受监管的数据删除，应使用合约规定的删除机制。

生产验收清单

• 明确选择原生模型还是合作伙伴模型。
• 重新核实当前模型 ID、端点、API 版本、架构、价格和停用状态。
• 每项输入均已取得同意并获授权，已在本地验证、完成哈希处理，且不含不必要的敏感数据。
• 提示词和参考素材策略与模式相符。
• 已审查试运行成本和受保护的完整载荷。
• 已提供确切的审批摘要、最高成本和独占任务键。
• 已禁用创建重试；创建状态不明时不得自动重新创建。
• 已持久化存储任务 ID；轮询能够正确处理 THROTTLED 和 CANCELLED。
• 轮询超时不会取消任务。DELETE 操作需单独取得破坏性操作审批。
• 在付费调用前执行内容审核；不得绕过被拦截的请求。
• 在过期前下载输出，验证主机、大小和类型，完整解码、计算哈希，并持久化存储。
• 由人工审核音频流、时长、fps、尺寸、连续性、伪影和验收标准。
• 在适用的 API 集成界面中显示“Powered by 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 公告
• 企业版第三方模型常见问题
• 使用权常见问题
• 安全与来源