Runway Video

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

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

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

#!/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()

$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

$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

{
  "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"}
}

{
  "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"}
}

{
  "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"}
}

{
  "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"}
}

#!/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")})