Runway Video

Runway Video

Use this skill for server-side integrations with the documented Runway Developer API at https://api.dev.runwayml.com. It covers Runway-native video models and makes partner models, still-image generation, the consumer web app, Recipes, Workflows, Characters, and third-party gateways explicit out-of-scope branches.

Research cutoff: 2026-07-10.

Evidence language

Interpret statements in this skill as follows:

• Fact — stated in a first-party Runway source or visible in the official OpenAPI-generated SDK.
• Claim — Runway marketing or policy language that may depend on plan or contract.
• Heuristic — an operational recommendation derived from the facts; validate it for the deployment.
• Unknown — not established by the reviewed first-party public material. Do not silently fill it in.

When a volatile fact affects money, safety, payload validity, or data governance, re-open the cited first-party page before production use.

Start with scope, consent, and an execution mode

Before designing a request, obtain all of the following:

1. The desired mode: text-to-video, image-to-video, video edit, or performance transfer.
2. Whether Runway-native models are required or a partner model is acceptable.
3. Output duration, dimensions, delivery format, audio expectation, and iteration budget.
4. Rights to every prompt, image, video, face, voice, performance, mark, and music input.
5. Explicit permission from each depicted or heard person. Runway's Usage Policy prohibits using another person's image, video, or audio without permission.
6. A disclosure/provenance plan when realistic synthetic media could mislead viewers.
7. Data classification, approved regions/subprocessors, retention requirement, and whether the organization's contract actually provides no-training or zero-data-retention terms.
8. One of these execution modes:
   • Plan/dry-run — default. Validate, price, and print an approval digest. Perform no paid API creation.
   • Approved create — only after an exact request digest, bounded maximum cost, and explicit environment gates match.
   • Resume — poll an already recorded task ID; never create a replacement.
   • Cancel/delete — a separate destructive operation with a separate approval. Never infer it from a timeout.

Refuse requests that lack necessary rights or consent, seek moderation evasion, impersonation, fraud, non-consensual intimate imagery, sexual content involving minors, or another prohibited category. Do not “fix” blocked content by obfuscating it.

Route the request

Current Runway-native video surface

Facts: gen4.5, gen4_turbo, aleph2, and act_two are in the current API model catalog. gen4_aleph is deprecated and scheduled to sunset on 2026-07-30. Do not start new work on it.

Native mode details

• Gen-4.5 text-to-video: ratios are 1280:720 and 720:1280. The request schema requires a non-empty prompt of at most 1,000 UTF-16 code units, a ratio, and an integer duration from 2 through 10.
• Gen-4.5 image-to-video: ratios are 1280:720, 720:1280, 1104:832, 832:1104, 1584:672, and 960:960. The current SDK schema accepts one first-frame image, either as a URI or an array item with position: "first".
• Gen-4 Turbo: image input is required. It supports the six ratios above and any duration from 2 through 10 seconds. It is the lower-cost native image-driven option.
• Aleph 2.0: input video must be 2–30 seconds, at most 30 fps, and no more than 1080p. Output preserves input resolution up to 1080p. The API accepts optional promptText, seed, contentModeration, targetAspectRatio, and up to five keyframes. Each keyframe uses a URI and either absolute seconds or normalized at; optional edit ranges require all keyframes to use ranges or none to use them.
• Act-Two: character is {type: "image"|"video", uri}, and reference is {type: "video", uri}. The driving performance is 3–30 seconds. The API also accepts bodyControl, expressionIntensity 1–5, ratio, seed, and contentModeration. Body/gesture transfer is intended for character-image inputs; a character video retains its own environmental/camera movement and does not offer the same gesture-control behavior.

Documented conflict: the Inputs page also lists Gen-4.5 portrait ratio 672:1584, but the reviewed current OpenAPI-generated SDK type omits it. Treat that ratio as unresolved rather than guessing; use one of the common values above or recheck the live API reference/SDK before approving 672:1584.

Partner models are a separate decision

Runway also exposes partner video models: seedance2, seedance2_fast, seedance2_mini, veo3, veo3.1, veo3.1_fast, happyhorse_1_0, and gemini_omni_flash. Magnific video upscaling is also a partner capability. They have different schemas, prices, audio behavior, moderation diagnostics, failure modes, subprocessors, and possibly different contractual controls.

Do not silently replace a native model with one of these. If the user chooses a partner model, re-research that model's current first-party Runway schema and provider-specific rights/data conditions, create a new cost formula, and obtain a new approval digest. Runway's enterprise FAQ says its enterprise terms and DPAs cover partner models and that providers have no-training commitments; that is not proof that every self-serve API account has the same terms.

Explicitly out of scope

• Still-image routes (/v1/text_to_image) are a staging option, not video generation. A still-image call is separately billable and requires its own skill/workflow and approval.
• The consumer web app, Explore Mode, Apps, Edit Studio UI, and web credits are not API contracts. Web-app credits and Developer API credits are separate.
• Recipes, published Workflows, real-time Characters, avatar videos, audio-only endpoints, and video upscaling are separate products/endpoints.
• Replicate, fal, ComfyUI wrappers, reverse-engineered endpoints, browser automation, and other gateways are not the Runway Developer API.

Build a shot specification before a prompt

Record a compact shot spec:

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"]

Then write a direct, positive prompt. For Gen-4.5 text-to-video, describe both what is visible and how it moves. For image-to-video, the image already establishes subject, composition, color, lighting, and style, so emphasize subject, camera, and environmental motion. Add one variable at a time during iteration.

Heuristics:

• Prefer one shot and one main action per short clip.
• Use concrete camera language: “locked camera,” “slow dolly in,” “handheld tracking,” or “overhead static frame.”
• Replace negatives with positive outcomes. Use “locked camera” instead of “no camera movement.”
• Do not ask the model to render critical legal copy or logos reliably; composite approved text and marks in post.
• Keep a fixed seed only when comparing prompt variants. A repeated seed is a similarity aid, not a determinism guarantee.

Example Gen-4.5 text 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. Natural commercial cinematography, crisp product silhouette, gentle background falloff.

For Aleph, use an action verb plus a targeted transformation: “Relight the room with cool moonlight while preserving the actor, wardrobe, camera motion, and set layout.” Start with the smallest edit. Use a keyframe when exact appearance matters; use promptText for motion or a transformation that is not visible in the keyframe.

For Act-Two, quality comes primarily from references rather than prose: one subject, visible face throughout, waist-up or closer, even lighting, no cuts, hands visible at the start if body control is enabled, and natural motion. Begin with expressionIntensity: 3. Higher values can add expression and artifacts; lower values can improve stability.

API and asset contracts

Every API request uses:

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

Keep the secret server-side. Never put it in source, logs, prompts, browser code, asset URLs, approval records, or output-download requests. Pin the API base URL; do not accept it from end users.

Input media

Runway accepts HTTPS URLs, data URIs, and runway:// ephemeral-upload URIs.

HTTPS inputs must use a hostname rather than an IP, respond to HEAD, return accurate Content-Type and Content-Length, and return 200 without redirects. Runway's fetch timeout is documented as 10 seconds. Supported image encodings are JPEG, PNG, and WebP; GIF is unsupported. Supported video containers/codecs are documented on the Inputs page.

Heuristic: use a private, exact-host allowlist and short-lived signed object URL when an approved object store already exists. Otherwise use the official ephemeral-upload API. Validate content locally before upload and bind its SHA-256 to the approval. Never let an end user supply an arbitrary fetch URL; that becomes an SSRF and data-exfiltration surface in your application even though Runway applies its own URL restrictions.

runway:// ephemeral uploads expire after 24 hours, are rate limited, require purchased credits, and are 512 bytes–200 MB. A failed multipart upload must not be retried; create a new upload session. An upload is stateful, so record the URI without putting the API secret or source bytes in logs.

Output media

A successful task returns an output array of one or more URLs. Runway says these URLs expire within 24–48 hours and must be downloaded into durable storage; do not expose them as product URLs. The official SDK type notes that retrieving the task again can return fresh URLs.

Treat output URLs as untrusted signed URLs:

• Never attach Authorization to them.
• Require HTTPS, no userinfo, default port, public DNS, an exact output-host allowlist, bounded redirects with every hop revalidated, bounded byte count, expected media type, and a container signature.
• Download into a private temporary file, compute SHA-256, run ffprobe, fully decode with ffmpeg -v error -i … -f null -, then atomically publish.
• Store task ID, request digest, model, endpoint, prompt hash or protected prompt, source hashes, approval, timestamps, output hash, codec/dimensions/fps/duration/audio-stream presence, and disclosure/rights records. Do not store signed query strings unless a security-reviewed incident workflow needs them.

The public API output page does not promise a single container/codec for every current model. A validator that supports only MP4 must fail closed on other output types rather than renaming them.

Price before creating

Developer credits are publicly priced at $0.01 each before applicable tax.

Use these formulas for native modes:

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

The Aleph and Act-Two ceiling formulas are conservative heuristics for approval when source duration is fractional. Public materials do not establish API rounding behavior for fractional seconds. Aleph's consumer Edit Studio says a ranged edit uses credits only for the range, but the reviewed API pricing page does not state the API's range-billing rule; price the full source duration unless the Developer API documentation or contract confirms otherwise.

Examples:

Moderated generations cost the same as successful generations. Do not use a paid Runway call as a moderation preflight. Pre-moderate in your application and include moderation failures in the budget.

Runway's public API documents tier-dependent concurrency, rolling 24-hour generation limits, and 30-day spend caps. Over concurrency, a task becomes THROTTLED and remains queued; over the daily create limit, creation returns 429. There is no general requests-per-minute cap documented, but this is not permission to create without your own queue, budget, and abuse controls.

Exact paid approval

A paid request must satisfy all of these at the moment of creation:

1. Dry-run produced a canonical request envelope containing endpoint, model, exact JSON payload, input SHA-256 values, local job key, output policy, current price source date, and a conservative maximum USD cost.
2. A human reviewed the non-redacted request in a protected interface.
3. The human approved that exact envelope digest, not merely “use Runway.”
4. SEND_RUNWAY_REQUEST=1 is set.
5. APPROVE_RUNWAY_PAID_CALL=YES is set.
6. RUNWAY_APPROVAL_SHA256 exactly matches the printed SHA-256.
7. RUNWAY_MAX_COST_USD is present and is at least the computed upper bound.
8. The create ledger acquired the job key exclusively before the POST.

Any change to prompt, model, endpoint, seed, duration, ratio, references, content-moderation setting, price table, or source hash invalidates approval.

Duplicate-create control and asynchronous lifecycle

Runway creation returns a task ID. The reviewed API reference and OpenAPI-generated SDK expose no documented idempotency-key parameter for these creation endpoints. Therefore:

• Create a local job record with exclusive-create semantics before the POST.
• Store a request digest and state: creating first.
• Send each create POST once. Disable automatic retries for create calls.
• Persist the returned task ID immediately and durably.
• If the client times out, disconnects, crashes, receives malformed success, or gets a 5xx after bytes may have been sent but before a task ID is stored, mark the job ambiguous. Do not create again automatically. Reconcile through provider support/usage records or accept a human-authorized possible duplicate under a new job key.
• If a task ID is known, resume GET /v1/tasks/{id}. Never replace it with a new generation merely because polling timed out.

The official SDK retries connection errors, 408, 409, 429, and 5xx twice by default. That is convenient for reads, but unsafe for a paid create without a provider idempotency contract. If using the SDK, instantiate the create client with max_retries=0; a separate read client may retry safe GETs.

Poll no more often than every five seconds. Add jitter and exponential backoff for retryable read errors. Recognize exactly:

CANCELLED uses two “l” characters in the official SDK schema.

A polling deadline or SDK wait timeout does not cancel a task. DELETE /v1/tasks/{id} is overloaded: it cancels PENDING, THROTTLED, or RUNNING tasks, but deletes other tasks and their persistent output data. A read-before-delete has a race if the task completes between calls. Never invoke DELETE automatically; require a separate explicit destructive approval that acknowledges possible terminal-task deletion. The public docs do not state the refund/charge rule for cancellation, so cancellation is not a cost-control guarantee.

Complete production reference

The following standard-library Python client implements Gen-4.5 text-to-video directly and accepts a validated source plan for native image-to-video, Aleph, or Act-Two. It is dry-run by default, binds exact approval to the paid payload, source evidence, price, output host/network policy, and governance evidence, uses an exclusive local ledger, never retries create, safely retries polling, refuses automatic cancellation, downloads without the API bearer token, and validates MP4 with ffprobe plus a full ffmpeg decode.

Python 3.11+, ffprobe, and ffmpeg are required. The companion source-plan builder after the payload templates supplies the other native modes without weakening this lifecycle.

#!/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": {
            "

…(truncated)