Minimax Hailuo Video

MiniMax Hailuo video

Build a reviewed request, obtain exact cost approval, create at most one paid task, then preserve the result and its evidence. Treat documentation gaps as unknowns, not defaults.

Start with the boundary

Choose one service region and keep its account, key, billing, host, and terms together:

The two platforms are separate service regions. A key or plan bought on one must not be assumed to work on the other. A 401 can be a wrong-region symptom. Use a pay-as-you-go API key for the examples below; a Token Plan/Subscription Key has separate quotas and model access.

Keep these products out of this workflow:

• hailuoai.video is the consumer Hailuo product with separate terms, privacy behavior, features, and commercial-use implications.
• Video Agent/template generation is a different template API. Its deprecated query endpoint and nine-hour link statement do not define general video generation.
• Image, speech, music, and text endpoints are separate modalities. General Hailuo video generation does not document an audio input, audio prompt, or generated-audio control.
• Gateways and resellers have their own schemas, retention, pricing, and support. Do not call them “the MiniMax API.”

Route by mode before choosing a model

Current first-party general video generation uses POST /v1/video_generation for four modes. The API overview highlights the Hailuo 2.3 family and Hailuo-02; detailed endpoint pages also enumerate older models.

MiniMax-Hailuo-2.3-Fast is I2V-only. Do not select it for T2V. Detailed pages describe older *-01* models as 720P/6 s, but current pay-as-you-go pricing omits them; verify availability and price in the correct console before any paid use. Do the same for S2V-01 rather than borrowing another model's price.

Current pay-as-you-go prices, checked 2026-07-10:

These are per generated video, not timeless constants. Recheck the region's official pricing page immediately before approval. Packages deduct units and raise RPM; Token Plans use plan quotas. Never turn a package's blended cost or another region's currency into a pay-as-you-go quote.

Build a request dossier

Before calling create, record:

• region, console/account identifier, key type, model, mode, resolution, duration, variant count, currency, unit price, and pricing-page retrieval time;
• exact prompt, prompt_optimizer, fast_pretreatment, and every non-default parameter;
• source file SHA-256, MIME type, byte length, dimensions, aspect ratio, provenance, license/consent, and any face/identity present;
• intended audience, jurisdiction, AI disclosure/watermark plan, retention class, and reviewer;
• an attempt ID generated locally before the POST.

For first/last-frame and I2V inputs, official constraints are JPG/JPEG, PNG, or WebP; under 20 MB; short edge over 300 px; aspect ratio from 2:5 through 5:2; public URL or Base64 data URL. The first image determines output aspect/resolution. In first/last mode, a differently sized last image is cropped to match the first. Pre-align composition, subject scale, lighting, and topology to avoid an implausible transition.

The S2V reference documents a different exact shape: "subject_reference": [{"type": "character", "image": ["https://authorized.example/reference.jpg"]}]. type is character; image is an array containing the authorized face-reference URL. Do not transpose I2V fields or assume I2V's media transport rules where the S2V page is silent.

A public source URL lets MiniMax fetch that asset and may expose it through URL logs or access controls. Prefer an appropriately scoped URL or bounded Base64 only when policy allows. Never accept arbitrary user URLs into your own downloader without SSRF controls. Hash the original asset; do not log Base64 bodies or signed download URLs.

Write prompts for this motion system

The prompt limit is 2,000 characters. Hailuo supports explicit camera commands on the documented 2.3, 02, and Director workflows:

[Truck left], [Truck right], [Pan left], [Pan right], [Push in], [Pull out], [Pedestal up], [Pedestal down], [Tilt up], [Tilt down], [Zoom in], [Zoom out], [Shake], [Tracking shot], [Static shot].

Commands in one bracket happen together; official guidance recommends at most three combined commands. Commands placed sequentially express a sequence. They are directional controls, not frame-accurate timecodes.

Use this compact prompt order:

1. Establish subject and initial composition.
2. Describe one physically coherent action arc.
3. State the dominant camera command or simple sequence.
4. Specify a visible end state and stable details that must persist.

For I2V, describe change from the supplied frame instead of contradicting its static appearance. For first/last-frame generation, describe a plausible bridge between the two known states. For a designed shot, set prompt_optimizer: false; otherwise the optimizer may rewrite wording or camera intent. prompt_optimizer defaults to true. fast_pretreatment defaults to false, requires the optimizer, and is documented for current Hailuo I2V models.

The following are craft heuristics, not API guarantees: one primary action and one dominant camera move usually fit a 6–10 second clip better than dense choreography; explicit end-state language improves reviewability; generated physics, faces, text, hands, occlusion, and identity should still be inspected frame by frame. MiniMax's claims about improved motion, expression, realism, and adherence are provider claims, not independent benchmark findings.

The general endpoint documents no seed, negative prompt, arbitrary aspect-ratio parameter, audio input, audio prompt, or audio-output switch. Do not invent them. Treat the presence and nature of an output audio track as unknown until the actual file is inspected; add separately licensed sound downstream when required.

Across the mode-specific references, model is required; T2V requires prompt, while I2V and first/last describe the text prompt as optional; prompts may be at most 2,000 characters. first_frame_image, last_frame_image, and subject_reference are required only by their respective modes. callback_url is optional. duration and resolution must follow the model/mode matrix above rather than generic CLI defaults. The current mainland-China T2V page additionally exposes aigc_watermark as a boolean defaulting to false; the current global T2V page does not.

Make the paid boundary explicit

Default to dry-run. Show the redacted request, exact one-video cost, variant multiplier, currency, authoritative pricing timestamp/evidence digest, source hashes, local job key, output destination, and exact-host download policy. Canonically hash that complete envelope. Require a human to approve the digest and a finite positive maximum in the same currency; amount-only approval is insufficient because it does not bind the model, prompt, source, region, or custody policy. If the combination has no current published price, stop with UNKNOWN and send the operator to the correct console or MiniMax support.

Before POSTing, acquire the job ledger with exclusive-create semantics and persist an intent-recorded entry. After a response, persist only an allowlisted result summary—task ID, provider code, status-message/body digests, and timestamps—not raw prompts, Base64, error bodies, or signed URLs. MiniMax documents no idempotency key for video creation. Therefore:

• Retry authenticated GET/poll/download operations with bounded backoff.
• Do not automatically retry a create POST.
• If create times out or the connection breaks before a task_id is captured, mark it create-outcome-unknown. Reconcile billing/console/support before authorizing a replacement task.
• Treat a 5xx, provider timeout/internal code, malformed envelope, or nominal success with an unusable task ID as equally ambiguous and persist that state before stopping.
• A replay is a new potentially billable video unless the earlier create is conclusively known not to exist.

HTTP success alone is not provider success. Require base_resp.status_code == 0 and a nonempty task_id. Preserve base_resp.status_code and status_msg for diagnosis. Common codes include 1001 timeout, 1002 rate limit, 1004 authentication, 1008 insufficient balance, 1026 input sensitive, 1027 output sensitive, 2013 invalid parameters, 2049 invalid key, and 2056 usage limit. Never evade a content-safety response.

The published default global video limit is currently 5 requests/minute across the Hailuo 2.3 and 02 series; packages publish higher limits. Treat this as account/plan-specific and recheck the console. Pace polls separately; the guide recommends a 10-second interval.

Check https://status.minimax.io/ during incidents, but do not infer healthy video generation merely from another listed component's status if video is not shown separately.

Complete dry-run-first client

This Python 3.11+ client covers currently priced T2V, I2V, and first/last-frame combinations in both regions. It does not execute S2V-01 or unpriced legacy models. It never retries create, acquires an exclusive attempt ledger before create, resumes known task IDs, checks MiniMax's envelope, polls every 10 seconds, downloads with an exact-host/public-DNS policy and no bearer token, and crash-resumably validates MP4 with ffprobe plus a full ffmpeg decode. Save it as hailuo_generate.py outside the skill directory. Python 3.11+, ffprobe, and ffmpeg are required.

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

HOSTS = {"global": "https://api.minimax.io", "cn": "https://api.minimaxi.com"}
PRICES = {
    "global": {"currency": "USD", "table": {
        ("MiniMax-Hailuo-2.3-Fast", "768P", 6): "0.19", ("MiniMax-Hailuo-2.3-Fast", "768P", 10): "0.32",
        ("MiniMax-Hailuo-2.3-Fast", "1080P", 6): "0.33", ("MiniMax-Hailuo-2.3", "768P", 6): "0.28",
        ("MiniMax-Hailuo-2.3", "768P", 10): "0.56", ("MiniMax-Hailuo-2.3", "1080P", 6): "0.49",
        ("MiniMax-Hailuo-02", "512P", 6): "0.10", ("MiniMax-Hailuo-02", "512P", 10): "0.15",
        ("MiniMax-Hailuo-02", "768P", 6): "0.28", ("MiniMax-Hailuo-02", "768P", 10): "0.56",
        ("MiniMax-Hailuo-02", "1080P", 6): "0.49"}},
    "cn": {"currency": "CNY", "table": {
        ("MiniMax-Hailuo-2.3-Fast", "768P", 6): "1.35", ("MiniMax-Hailuo-2.3-Fast", "768P", 10): "2.25",
        ("MiniMax-Hailuo-2.3-Fast", "1080P", 6): "2.31", ("MiniMax-Hailuo-2.3", "768P", 6): "2.00",
        ("MiniMax-Hailuo-2.3", "768P", 10): "4.00", ("MiniMax-Hailuo-2.3", "1080P", 6): "3.50",
        ("MiniMax-Hailuo-02", "512P", 6): "0.60", ("MiniMax-Hailuo-02", "512P", 10): "1.00",
        ("MiniMax-Hailuo-02", "768P", 6): "2.00", ("MiniMax-Hailuo-02", "768P", 10): "4.00",
        ("MiniMax-Hailuo-02", "1080P", 6): "3.50"}}
}
MODELS = {
    "text": {"MiniMax-Hailuo-2.3", "MiniMax-Hailuo-02"},
    "image": {"MiniMax-Hailuo-2.3", "MiniMax-Hailuo-2.3-Fast", "MiniMax-Hailuo-02"},
    "first-last": {"MiniMax-Hailuo-02"},
}
MIME = {".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png", ".webp": "image/webp"}
MAX_JSON = 256 * 1024
MAX_VIDEO = 250 * 1024 * 1024
MAX_TOOL_OUTPUT = 512 * 1024
NONTERMINAL = {"Preparing", "Queueing", "Processing"}

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, http_status=None, body_sha256=None):
        super().__init__(kind); self.kind = kind; self.http_status = http_status; self.body_sha256 = 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 image_dimensions(data, suffix):
    if suffix == ".png" and data[:8] == b"\x89PNG\r\n\x1a\n" and data[12:16] == b"IHDR":
        return int.from_bytes(data[16:20], "big"), int.from_bytes(data[20:24], "big")
    if suffix in {".jpg", ".jpeg"} and data[:2] == b"\xff\xd8":
        i = 2
        while i + 9 < len(data):
            if data[i] != 0xff: i += 1; continue
            marker = data[i + 1]; i += 2
            if marker in {0xd8, 0xd9} or 0xd0 <= marker <= 0xd7: continue
            if i + 2 > len(data): break
            length = int.from_bytes(data[i:i + 2], "big")
            if marker in {0xc0, 0xc1, 0xc2, 0xc3, 0xc5, 0xc6, 0xc7, 0xc9, 0xca, 0xcb, 0xcd, 0xce, 0xcf}:
                return int.from_bytes(data[i + 5:i + 7], "big"), int.from_bytes(data[i + 3:i + 5], "big")
            if length < 2: break
            i += length
    if suffix == ".webp" and data[:4] == b"RIFF" and data[8:12] == b"WEBP":
        kind = data[12:16]
        if kind == b"VP8X" and len(data) >= 30:
            return 1 + int.from_bytes(data[24:27], "little"), 1 + int.from_bytes(data[27:30], "little")
        if kind == b"VP8L" and len(data) >= 25 and data[20] == 0x2f:
            b0, b1, b2, b3 = data[21:25]
            return 1 + b0 + ((b1 & 0x3f) << 8), 1 + (b1 >> 6) + (b2 << 2) + ((b3 & 0x0f) << 10)
        if kind == b"VP8 " and len(data) >= 30 and data[23:26] == b"\x9d\x01\x2a":
            return int.from_bytes(data[26:28], "little") & 0x3fff, int.from_bytes(data[28:30], "little") & 0x3fff
    raise SystemExit("Image signature or dimensions are invalid/unsupported")

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

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

def read_json(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 media(path_text):
    path = pathlib.Path(path_text)
    if not path.is_file() or path.suffix.lower() not in MIME or path.stat().st_size >= 20 * 1024 * 1024:
        raise SystemExit("Image must be JPG/JPEG, PNG, or WebP and strictly under 20 MB")
    data = path.read_bytes()
    width, height = image_dimensions(data, path.suffix.lower())
    ratio = width / height
    if min(width, height) <= 300 or not 0.4 <= ratio <= 2.5:
        raise SystemExit("Image short edge must exceed 300 px and aspect ratio must be within 2:5–5:2")
    bounded_process(["ffmpeg", "-nostdin", "-v", "error", "-i", str(path), "-f", "null", "-"], 60)
    return {
        "wire": f"data:{MIME[path.suffix.lower()]};base64,{base64.b64encode(data).decode('ascii')}",
        "evidence": {"path": str(path.resolve()), "sha256": hashlib.sha256(data).hexdigest(),
                     "bytes": len(data), "mime": MIME[path.suffix.lower()],
                     "width": width, "height": height, "aspect_ratio": ratio}
    }

def api_json(method, url, key, body=None, retries=0):
    encoded = None if body is None else canonical(body)
    for attempt in range(retries + 1):
        req = urllib.request.Request(url, data=encoded, method=method,
            headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json", "Accept": "application/json"})
        try:
            with NO_REDIRECT.open(req, timeout=40) as response:
                raw = response.read(MAX_JSON + 1)
            if len(raw) > MAX_JSON: raise APIProblem("response-too-large")
            try: result = json.loads(raw)
            except json.JSONDecodeError: raise APIProblem("invalid-json", response.status, sha256_bytes(raw))
            if not isinstance(result, dict): raise APIProblem("non-object-json", response.status, sha256_bytes(raw))
            return result
        except urllib.error.HTTPError as exc:
            raw = exc.read(MAX_JSON + 1)
            problem = APIProblem("http-error", exc.code, sha256_bytes(raw))
            if attempt == retries or exc.code not in {408, 429, 500, 502, 503, 504}: 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(8, 2 ** attempt) * (1 + random.random() * 0.25))
    raise AssertionError("unreachable")

def provider_result(result):
    base = result.get("base_resp")
    if not isinstance(base, dict) or not isinstance(base.get("status_code"), int):
        raise APIProblem("missing-provider-envelope", body_sha256=sha256_bytes(canonical(result)))
    return base["status_code"], sha256_bytes(str(base.get("status_msg", "")).encode("utf-8"))

def exact_hosts(raw):
    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("Output hosts must be exact DNS names")
        try: ipaddress.ip_address(host)
        except ValueError: pass
        else: raise SystemExit("Output host policy rejects IP literals")
        hosts.append(host)
    if not hosts: raise SystemExit("Provide at least one reviewed exact --output-host")
    return tuple(sorted(set(hosts)))

def validate_output_url(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:
        raise RuntimeError("Output URL must be HTTPS with hostname, no userinfo, and no fragment")
    if parsed.port not in (None, 443): raise RuntimeError("Output URL must use default HTTPS port")
    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 = []
    for item in socket.getaddrinfo(host, 443, type=socket.SOCK_STREAM):
        address = ipaddress.ip_address(item[4][0])
        if not address.is_global: raise RuntimeError("Output host resolves to a non-public address")
        addresses.append(str(address))
    if not addresses: raise RuntimeError("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, 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 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:
        header = stream.read(32)
        if len(header) < 12 or header[4:8] != b"ftyp": raise RuntimeError("Artifact lacks ISO-BMFF 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 safe_download(url, destination, allowed_hosts, ledger_path, record):
    destination = pathlib.Path(destination); destination.parent.mkdir(parents=True, exist_ok=True)
    stage = destination.with_name("." + destination.name + ".hailuo-stage")
    expected = record.get("artifact_staged") or record.get("artifact")
    if destination.exists():
        if not isinstance(expected, dict): raise RuntimeError("Existing artifact has no durable expected digest")
        artifact = verify_mp4(destination)
        if (artifact["sha256"], artifact["bytes"]) != (expected.get("sha256"), expected.get("bytes")):
            raise RuntimeError("Existing artifact does not match durable evidence")
        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 does not match durable evidence")
        os.replace(stage, destination); fsync_directory(destination.parent); return artifact
    host, target, addresses = validate_output_url(url, allowed_hosts)
    url_hash = sha256_bytes(url.encode("utf-8"))
    if record.get("download_url_sha256") not in (None, url_hash): raise RuntimeError("Download URL changed during an incomplete transfer")
    record.update(state="downloading", download_url_sha256=url_hash, download_host=host, download_started_at=record.get("download_started_at") or utc_now())
    atomic_json(ledger_path, record)
    if stage.exists(): stage.unlink()
    connection = PinnedHTTPSConnection(host, addresses[0], timeout=90)
    try:
        connection.request("GET", target, headers={"Host": host, "Accept": "video/mp4", "User-Agent": "hailuo-artifact-client/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 or 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 out:
                while True:
                    chunk = response.read(1024 * 1024)
                    if not chunk: break
                    total += len(chunk)
                    if total > MAX_VIDEO: raise RuntimeError("Download exceeded byte cap")
                    out.write(chunk)
                out.flush(); os.fsync(out.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())
    atomic_json(ledger_path, record)
    os.replace(stage, destination); fsync_directory(destination.parent)
    return artifact

def main():
    p = argparse.ArgumentParser()
    p.add_argument("--region", choices=HOSTS, required=True)
    p.add_argument("--mode", choices=MODELS, required=True)
    p.add_argument("--model", required=True); p.add_argument("--prompt", required=True)
    p.add_argument("--resolution", choices=["512P", "768P", "1080P"], required=True)
    p.add_argument("--duration", type=int, choices=[6, 10], required=True)
    p.add_argument("--first-frame"); p.add_argument("--last-frame")
    p.add_argument("--optimizer", action=argparse.BooleanOptionalAction, default=False)
    p.add_argument("--fast-pretreatment", action="store_true")
    p.add_argument("--execute", action="store_true")
    p.add_argument("--job-key", required=True)
    p.add_argument("--output-host", required=True, help="Comma-separated reviewed exact DNS hostnames")
    p.add_argument("--pricing-checked-at", required=True, help="Timezone-aware ISO-8601 time of the live regional price check")
    p.add_argument("--pricing-evidence-sha256", required=True)
    p.add_argument("--rights-sha256", required=True); p.add_argument("--moderation-sha256", required=True)
    p.add_argument("--qa-plan-sha256", required=True)
    p.add_argument("--approval-sha256")
    p.add_argument("--max-cost"); p.add_argument("--max-cost-currency", choices=["USD", "CNY"])
    p.add_argument("--show-full-plan", action="store_true")
    p.add_argument("--ledger", default="hailuo-attempt.json")
    p.add_argument("--output", default="hailuo-output.mp4")
    a = p.parse_args()
    if not a.job_key or any(not (character.isalnum() or character in "-_") for character in a.job_key):
        raise SystemExit("job-key accepts letters, digits, hyphen, and underscore")
    if a.model not in MODELS[a.mode]: raise SystemExit("Model is not valid for the selected mode")
    if len(a.prompt) > 2000: raise SystemExit("Prompt exceeds 2,000 characters")
    if a.fast_pretreatment and not a.optimizer: raise SystemExit("fast_pretreatment requires optimizer")
    if a.mode in {"image", "first-last"} and not a.first_frame: raise SystemExit("First frame required")
    if a.mode == "first-last" and not a.last_frame: raise SystemExit("Last frame required")
    key = (a.model, a.resolution, a.duration)
    unit = PRICES[a.region]["table"].get(key)
    if unit is None: raise SystemExit("UNKNOWN current price/combination; verify in the correct MiniMax console")
    if a.model == "MiniMax-Hailuo-02" and a.mode == "text" and a.resolution == "512P":
        raise SystemExit("512P pricing/availability is documented for Hailuo-02 I2V, not T2V")
    if a.mode == "first-last" and a.resolution == "512P":
        raise SystemExit("The first/last-frame endpoint explicitly does not support 512P")
    request = {"model": a.model, "prompt": a.prompt, "resolution": a.resolution,
               "duration": a.duration, "prompt_optimizer": a.optimizer}
    evidence = {}
    if a.mode in {"image", "first-last"}:
        item = media(a.first_frame); request["first_frame_image"] = item["wire"]; evidence["first_frame"] = item["evidence"]
    if a.mode == "first-last":
        item = media(a.last_frame); request["last_frame_image"] = item["wire"]; evidence["last_frame"] = item["evidence"]
    if a.fast_pretreatment: request["fast_pretreatment"] = True
    currency = PRICES[a.region]["currency"]
    output_hosts = exact_hosts(a.output_host)
    for name, value in (("pricing evidence", a.pricing_evidence_sha256), ("rights", a.rights_sha256),
                        ("moderation", a.moderation_sha256), ("QA plan", a.qa_plan_sha256)):
        if len(value) != 64 or any(character not in "0123456789abcdefABCDEF" for character in value):
            raise SystemExit(f"{name} must be a SHA-256 digest")
    redacted = dict(request)
    for field in ("first_frame_image", "last_frame_image"):
        if field in redacted: redacted[field] = f"<base64 redacted; sha256={evidence[field.split('_image')[0]].get('sha256', 'see evidence')}>"
    redacted["prompt"] = "sha256:" + sha256_bytes(a.prompt.encode("utf-8"))
    request_hash = sha256_bytes(canonical(request))
    output_policy = {"exact_hosts": list(output_hosts), "https": True, "redirects": "reject",
                     "reject_ip_literals": True, "public_dns": True,
                     "dns_pinning": "connect-IP-with-TLS-SNI-and-host-certificate",
                     "max_bytes": MAX_VIDEO, "container": "mp4", "full_decode": True}
    envelope = {"schema_version": 1, "job_key": a.job_key, "region": a.region, "api_origin": HOSTS[a.region],
                "endpoint": "/v1/video_generation", "mode": a.mode, "model": a.model,
                "request_sha256": request_hash, "redacted_request": redacted, "source_evidence": evidence,
                "output": str(pathlib.Path(a.output).resolve()), "output_policy": output_policy,
                "price": {"amount": unit, "currency": currency, "checked_at": a.pricing_checked_at,
                          "evidence_sha256": a.pricing_evidence_sha256.lower()},
                "governance": {"rights_sha256": a.rights_sha256.lower(),
                               "moderation_sha256": a.moderation_sha256.lower(),
                               "qa_plan_sha256": a.qa_plan_sha256.lower()}}
    approval = sha256_bytes(canonical(envelope))
    print(json.dumps({"approval_sha256": approval, "plan": envelope}, indent=2, ensure_ascii=False))
    if a.show_full_plan: print(json.dumps({"protected_payload": request}, indent=2, ensure_ascii=False))
    print(f"DRY RUN: one task, exact maximum {currency} {unit}; no network request sent", file=sys.stderr)
    if not a.execute: return
    if a.approval_sha256 != approval: raise SystemExit("Exact approval digest mismatch")
    try: maximum = Decimal(a.max_cost or "")
    except InvalidOperation: raise SystemExit("max-cost must be a decimal")
    if not maximum.is_finite() or maximum <= 0 or maximum < Decimal(unit):
        raise SystemExit("max-cost must be finite, positive, and at least the quoted amount")
    if a.max_cost_currency != currency: raise SystemExit("max-cost currency does not match the selected region")
    try: checked_at = datetime.fromisoformat(a.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 a timezone")
    age = datetime.now(timezone.utc) - checked_at.astimezone(timezone.utc)
    if age.total_seconds() < -300 or age.total_seconds() > 86400:
        raise SystemExit("Live regional pricing evidence must be no more than 24 hours old")
    api_key = os.environ.get("MINIMAX_API_KEY")
    if not api_key: raise SystemExit("Set MINIMAX_API_KEY only in the environment")
    ledger_path = pathlib.Path(a.ledger)
    if ledger_path.exists():
        intent = read_json(ledger_path)
        if intent.get("request_sha256") != request_hash or intent.get("approval_sha256") != approval:
            raise SystemExit("Existing job ledger belongs to a different approved request")
        task_id = intent.get("task_id")
        if not task_id: raise SystemExit(f"Ledger state {intent.get('state')!r} has no task ID; never replay this job key")
    else:
        intent = {"schema_version": 1, "state": "intent-recorded", "created_at": utc_now(),
                  "job_key": a.job_key, "request_sha256": request_hash, "approval_sha256": approval,
                  "region": a.region, "host": HOSTS[a.region], "mode": a.mode, "model": a.model,
                  "redacted_request": redacted, "source_evidence": evidence, "output_policy": output_policy,
                  "price": envelope["price"], "governance": envelope["governance"]}
        try: exclusive_json(ledger_path, intent)
        except FileExistsError: raise SystemExit("Another process acquired this job ledger")
        intent.update(state="create-started", create_started_at=utc_now()); atomic_json(ledger_path, intent)
        try: created = api_json("POST", HOSTS[a.region] + "/v1/video_generation", api_key, request, retries=0)
        except APIProblem as exc:
            intent.update(state="create-outcome-unknown", create_outcome_at=utc_now(), error_kind=exc.kind,
                          http_status=exc.http_status, error_body_sha256=exc.body_sha256, replay_allowed=False)
            atomic_json(ledger_path, intent)
            raise SystemExit("Create outcome unknown; do not replay until billing/console/support is reconciled")
        try: provider_code, provider_message_sha256 = provider_result(created)
        except APIProblem as exc:
            intent.update(state="create-outcome-unknown", create_outcome_at=utc_now(), error_kind=exc.kind,
                          response_sha256=exc.body_sha256, replay_allowed=False); atomic_json(ledger_path, intent)
            raise SystemExit("Create success envelope is unusable; do not replay")
        if provider_code != 0:
            uncertain = provider_code in {1001, 1024, 1033}
            intent.update(state="create-outcome-unknown" if uncertain else "create-rejected",
                          create_outcome_at=utc_now(), provider_status_code=provider_code,
                          provider_status_message_sha256=provider_message_sha256, replay_allowed=False)
            atomic_json(ledger_path, intent)
            raise SystemExit("MiniMax rejected or could not confirm create; this job key cannot be replayed")
        task_id = created.get("task_id")
        if not isinstance(task_id, (str, int)) or not str(task_id).isdigit():
            intent.update(state="create-outcome-unknown", create_outcome_at=utc_now(),
                          error_kind="unusable-success-task-id", response_sha256=sha256_bytes(canonical(created)),
                          replay_allowed=False); atomic_json(ledger_path, intent)
            raise SystemExit("Create returned an unusable task ID; do not replay")
        task_id = str(task_id)
        intent.update(state="task-created", task_id=task_id, submitted_at=utc_now(),
                      provider_status_code=0, provider_status_message_sha256=provider_message_sha256)
        atomic_json(ledger_path, intent)
    deadline = time.monotonic() + 30 * 60
    while time.monotonic() < deadline:
        query = HOSTS[a.region] + "/v1/query/video_generation?" + urllib.parse.urlencode({"task_id": task_id})
        try: status = api_json("GET", query, api_key, retries=3)
        except APIProblem as exc:
            intent.update(poll_error_kind=exc.kind, poll_http_status=exc.http_status,
                          poll_error_body_sha256=exc.body_sha256); atomic_json(ledger_path, intent); raise SystemExit("Polling failed; resume this task ID later")
        provider_code, provider_message_sha256 = provider_result(status)
        if provider_code != 0:
            intent.update(poll_provider_status_code=provider_code,
                          poll_provider_status_message_sha256=provider_message_sha256); atomic_json(ledger_path, intent)
            raise SystemExit("MiniMax query returned a provider error; resume after diagnosis")
        if status.get("task_id") not in (None, task_id, int(task_id)):
            raise SystemExit("Query task ID mismatch")
        name = status.get("status")
        intent.update(state="polling", last_status=name, last_polled_at=utc_now()); atomic_json(ledger_path, intent)
        if name == "Success":
            file_id = status.get("file_id")
            if not isinstance(file_id, (str, int)) or not str(file_id).isdigit(): raise RuntimeError("Success returned invalid file_id")
            file_id = str(file_id)
            retrieve = HOSTS[a.region] + "/v1/files/retrieve?" + urllib.parse.urlencode({"file_id": file_id})
            retrieved = api_json("GET", retrieve, api_key, retries=3)
            retrieve_code, retrieve_message_sha256 = provider_result(retrieved)
            if retrieve_code != 0: raise RuntimeError("MiniMax retrieve returned a provider error")
            file_record = retrieved.get("file") or {}
            if str(file_record.get("file_id")) not in {"None", file_id}: raise RuntimeError("Retrieve file ID mismatch")
            url = file_record.get("download_url") or retrieved.get("download_url")
            if not url: raise RuntimeError("Retrieve returned no download_url")
            artifact = safe_download(url, a.output, set(output_hosts), ledger_path, intent)
            expected_dimensions = (status.get("video_width"), status.get("video_height"))
            actual_dimensions = (artifact["qa"].get("width"), artifact["qa"].get("height"))
            try: actual_duration = Decimal(s

…(truncated)