Minimax Hailuo Video

#!/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(str(artifact["qa"].get("duration")))
            except InvalidOperation: actual_duration = Decimal("NaN")
            qa_state = "automated-passed"
            if expected_dimensions != actual_dimensions or not actual_duration.is_finite() or abs(actual_duration - Decimal(a.duration)) > Decimal("1"):
                qa_state = "quarantined-metadata-mismatch"
            intent.update(state="artifact-downloaded" if qa_state == "automated-passed" else "artifact-quarantined",
                          file_id=file_id, retrieve_provider_message_sha256=retrieve_message_sha256,
                          artifact=dict(artifact, path=str(pathlib.Path(a.output).resolve())),
                          expected_dimensions={"width": expected_dimensions[0], "height": expected_dimensions[1]},
                          qa_state=qa_state, human_review="pending", stored_at=utc_now())
            atomic_json(ledger_path, intent); print(json.dumps({"artifact": intent["artifact"], "qa_state": qa_state}, indent=2)); return
        if name == "Fail":
            intent.update(state="generation-failed", terminal_at=utc_now(),
                          failure_sha256=sha256_bytes(str(status.get("error_message", "")).encode("utf-8")))
            atomic_json(ledger_path, intent); raise SystemExit("MiniMax generation failed")
        if name not in NONTERMINAL: raise RuntimeError("Unknown task status; fail closed")
        time.sleep(10)
    intent.update(state="poll-deadline-reached", deadline_at=utc_now()); atomic_json(ledger_path, intent)
    raise SystemExit("Polling deadline reached; task may still be running—resume query, do not recreate")

if __name__ == "__main__": main()

python hailuo_generate.py --region global --mode text --model MiniMax-Hailuo-2.3 --resolution 768P --duration 6 --job-key kite-v1 --output-host '<reviewed-exact-download-host>' --pricing-checked-at '2026-07-10T12:00:00Z' --pricing-evidence-sha256 '<64-hex-digest>' --rights-sha256 '<64-hex-digest>' --moderation-sha256 '<64-hex-digest>' --qa-plan-sha256 '<64-hex-digest>' --prompt "A red paper kite rises above a quiet salt flat. [Tracking shot] The camera follows low and steady; the kite settles centered against the horizon."

export MINIMAX_API_KEY='replace-in-your-shell-only'
python hailuo_generate.py <the-exact-unchanged-dry-run-arguments> --execute --approval-sha256 '<printed-digest>' --max-cost 0.28 --max-cost-currency USD