Xai Grok Imagine Video

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

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

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

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

NO_REDIRECT = urllib.request.build_opener(NoRedirect)

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

if __name__ == "__main__":
    main()

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

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

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