interface:
  display_name: "短剧生产"
  short_description: "确认后安全执行图片、视频和语音生产任务，并保留可审计运行记录"
  default_prompt: "Use $short-drama-produce to preview a bounded media job, request explicit confirmation, then execute it safely."

# Adapter Contract

## Job file

```json
{
  "schema_version": "1.0",
  "job_id": "EP001-MOTION001-video",
  "modality": "video",
  "adapter": "studio-video",
  "prompt": "the exact copyable prompt from MOTION-EP001-001",
  "source": "剧集/EP001/视频提示词.md",
  "source_entry": "MOTION-EP001-001",
  "reference_bindings": [
    {
      "slot_id": "REF-HERO",
      "order": 1,
      "path": "输入/approved-character-reference.png",
      "label": "女主定妆照",
      "role": "reference_image",
      "may_control": ["身份", "造型"],
      "must_not_control": ["构图", "动作"]
    }
  ],
  "references": ["输入/approved-character-reference.png"],
  "outputs": ["剧集/EP001/制作成果/video/SHOT-EP001-001.mp4"],
  "parameters": {"duration": 5, "ratio": "9:16", "prompt_language": "zh-CN"},
  "overwrite": false
}
```

- `modality`: `image`, `video`, `tts`, or `music`. `tts` is one bounded spoken
  utterance; `music` is a separately accepted timeline-level cue or song and
  must not be smuggled into every shot's video job.
- `source`: optional current project text/spec that owns the prompt.
- `source_entry`: for a creator-first job, the exact uppercase H2 ID inside a
  canonical `剧集|episodes/<EP>/` creator document, matched to that document:
  `图片提示词.md` takes `IMG-*` (image), `视频提示词.md` takes `MOTION-*` (video),
  and `分镜.md` takes `SHOT-*` (image, whose prompt is the shot's
  `### 冻结关键帧提示词` body rather than `### 可复制提示词`). A new image/video job pointing to either canonical filename must
  provide the matching selector; an arbitrary Markdown file cannot impersonate a
  creator source. `prepare` selects that section and requires `prompt` to exactly
  equal its copyable prompt after Markdown quote markers are removed.
- `reference_bindings`: zero to sixteen ordered semantic bindings. Every entry has
  exactly `slot_id`, contiguous `order`, project-relative `path`, Chinese `label`,
  non-empty `role`, and non-empty `may_control` / `must_not_control` lists. When
  `source_entry` is present, these fields must exactly match that entry's
  `参考` (IMG) or `输入参考图` (SHOT/MOTION) declaration (except `role`, which is
  production metadata). Allowed and prohibited scopes may not overlap.
  The creator declaration also carries a `用途` — what this picture decides in
  this shot, from the closed set in the storyboard skill. `role` is its
  production-side translation and is chosen per provider, so the two are not
  compared; a binding whose `role` contradicts the declared `用途` is a defect
  the creator document, not this schema, is the authority on. This schema keeps
  `role` free-form because an external adapter names its own inputs, but the
  bundled video adapters read it as the provider's own role and accept only
  their published vocabulary: `first_frame`, `last_frame`, `reference_image`,
  `reference_video`, `reference_audio` for MiniMax, and the three `reference_*`
  values for Seedance. A video job carrying references without bindings fails
  closed there rather than having a role guessed for it.
- `references`: zero to sixteen current project files actually sent to production.
  It may be omitted when `reference_bindings` is present, in which case the paths
  are derived in binding order. If both are present, they must match exactly.
- `outputs`: one to sixteen unique paths rooted at top-level `production/` or
  `剧集|episodes/<EP>/制作成果|production/`; extensions must match the modality.
  A nested directory merely named `production` does not grant write access to
  protected input or delivery trees.
- `parameters`: provider-neutral public settings only. Image/video jobs carry the resolved
  `prompt_language` so a bundled compiler writes its appended reference contract in the same
  language as the copyable prompt; this execution-only setting is not forwarded as a provider
  request field. Secret-like keys are rejected.
- `overwrite`: must be explicitly true to replace an existing result.

`prepare` records internal digests of source/reference bytes and returns the exact confirmation phrase. Callers never
calculate those digests. A changed input requires prepare and confirmation again.
Jobs already prepared and stored before this contract remain readable under their
original fingerprint. New non-creator structured jobs remain valid without
`source_entry` and `reference_bindings`; they still receive the existing
path/digest/confirmation checks. A `music` job may continue to use the timeline
music section of canonical `视频提示词.md` without pretending that section is a
`MOTION-*` entry; other new image/video jobs using canonical creator paths must use
their matching selector.

## Adapter config

Keep this file outside the project:

```json
{
  "adapters": {
    "studio-image": {
      "command": ["python3", "/opt/studio/image_adapter.py"],
      "timeout_seconds": 600
    }
  }
}
```

`command` is an argv array, never a shell string. Timeout is 1–3600 seconds. Do not put credentials in this file;
let the adapter read its environment or operating-system credential store.

## Capability sources

Use the provider's current API reference for roles, limits and supported combinations.
A relay can expose a different subset, reject inputs itself or transform requests.
Record the tested execution path and distinguish its observations from the native API contract.

## Adapter stdin

The adapter receives one UTF-8 JSON document as raw stdin bytes. Read the
binary stream (for example `json.load(sys.stdin.buffer)`) rather than relying
on the machine locale. The document contains the confirmed job plus:

- `run_id`: unique attempt ID;
- `project_root`: local absolute path to a private, run-scoped snapshot containing
  the exact confirmed `source` and `references`. It is not the live creator
  project and is deleted after the attempt.
- `output_root`: empty private, run-scoped staging directory owned by the
  production tool. Every adapter output source must be a direct regular-file
  child of this directory. The whole directory is deleted after success or
  failure.
- `handle_path`: absolute path where the adapter records the provider's task id
  **as soon as it has one, before it starts polling** — write
  `{"provider_job_id": "<id>"}`, atomically. Unlike `output_root` this path
  survives the attempt.
- `collect_provider_job_id`: present only on a collect call. When it is set the
  adapter must **not submit anything**; it polls and downloads that existing
  task and returns its outputs as usual.

## Recovering a submitted task

Record the provider task ID before polling so an interrupted task can be collected without resubmission.
A local handle-write failure is non-fatal after submission. The tool copies the handle into the attempt
record on success or failure; `audit` reports unfinished tasks as `orphaned_provider_job` with
`action: collect_before_retry`.

`production_tool.py collect` retrieves the existing task without new submission or confirmation.

It may translate provider-neutral parameters into its chosen SDK/API. Optional
provider adapters under `scripts/` document and implement known translations;
the project job remains provider-neutral, and adapter selection, model access,
polling and credentials stay in the external runtime configuration.

The confirmed document includes `source_entry` and `reference_bindings`. Bundled
image/video compilers append an ordered reference contract to the provider prompt so the Chinese
label, role, allowed controls, and prohibited controls survive the handoff. Its prose follows
`parameters.prompt_language`; model-native compilers also use that model's actual reference tokens,
such as Seedance `@图片1` or H3 `<Picture 1>`, rather than adding a second generic numbering scheme.
External adapters must preserve equivalent semantics
or reject the job; silently reducing the input to an unlabelled file list is invalid.

## Adapter stdout

On success, write one bounded UTF-8 JSON object to stdout. Portable adapters
should ASCII-escape JSON strings so Windows locale settings cannot corrupt
project paths:

```json
{
  "outputs": [
    {
      "target": "剧集/EP001/制作成果/images/SHOT001.png",
      "source": "/temporary/adapter/result.png"
    }
  ],
  "provider_job_id": "optional-public-id"
}
```

Targets must appear in exactly the confirmed order. Sources must be direct
regular-file children of `output_root`, not symlinks. The tool opens each source
without following links, copies the pinned bytes into the project, records
size/media type/checksum, and removes staging. It never stores adapter
stdout/stderr or environment values.

On a provider failure, an adapter may write only this whitelisted evidence to stdout:

```json
{
  "error": {
    "provider": "studio-image",
    "category": "rate_limit",
    "code": "rate_limit_exceeded",
    "http_status": 429,
    "request_id": "request_safe_123",
    "retryable": true
  }
}
```

Never include provider messages, response bodies, prompts, paths or credentials.
The production run record keeps only validated fields; malformed failure output
is replaced with a generic adapter exit code.

A nonzero exit, timeout, malformed response or mismatched output marks the run failed. Because the adapter may have
submitted paid work before failing locally, confirmation is consumed as soon as execution starts; retry only after a
new creator confirmation. A job with an unresolved `running` attempt cannot be prepared, confirmed, or run again;
wait for its terminal record or investigate the interrupted attempt first.

## Operational reconciliation

`production_tool.py audit <project>` reads only local production metadata and the
current output bytes. It reports terminal failures, retryable failures, recovered
jobs, unresolved running attempts, repeated content fingerprints, and missing or
changed outputs. Terminal state, recovery, and repeated output claims are ordered
by validated completion time rather than start time. For paths written more than
once, the latest completed successful claim is authoritative for this operational
check. Only attempts bound to the current stored job fingerprint can claim its
current outputs; older fingerprints remain visible as `superseded` attempt history.
A successful record must declare exactly the current job's output set and media
types before any output claim is trusted. Jobs, confirmations, run records, and the
production lock use pinned no-follow directories where directory FDs are available,
and reparse/identity checks on the portable fallback. A linked parent chain fails
closed instead of redirecting state outside the project.

The audit deliberately returns `quality_verdict: not_assessed`. A provider success,
retry recovery, file presence, size, or checksum proves execution/custody only; it
does not prove identity, performance, continuity, lip-sync, mix, edit, or audience
quality. Those claims require an authorized observation and a separate review.

#!/usr/bin/env python3
"""Offline adapter fixture for image, video, TTS, and music production tests."""

from __future__ import annotations

import argparse
import base64
import json
import struct
import sys
import wave
from pathlib import Path

MINIMUM_PYTHON = (3, 9)
if sys.version_info < MINIMUM_PYTHON:
    raise SystemExit(
        "short-drama-produce fixture needs Python {}.{} or newer".format(
            *MINIMUM_PYTHON
        )
    )

PNG = base64.b64decode(
    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII="
)


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--fail", action="store_true")
    # A real video provider bills at submission and can be interrupted at any
    # point afterwards. `--submit-then-fail` reproduces exactly that: record the
    # handle the way a live adapter does, then die before returning anything.
    parser.add_argument("--submit-then-fail", action="store_true")
    args = parser.parse_args()
    job = json.load(sys.stdin.buffer)
    collecting = job.get("collect_provider_job_id")
    if args.submit_then_fail and not collecting:
        destination = job.get("handle_path")
        if isinstance(destination, str) and destination:
            path = Path(destination)
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_text(
                json.dumps({"provider_job_id": "fixture-task-1"}), encoding="utf-8"
            )
        return 7
    if args.fail:
        return 7
    directory_raw = job.get("output_root")
    if not isinstance(directory_raw, str):
        return 8
    directory = Path(directory_raw)
    if not directory.is_absolute() or not directory.is_dir() or directory.is_symlink():
        return 8
    outputs = []
    for index, target in enumerate(job["outputs"]):
        suffix = Path(target).suffix.lower()
        path = directory / f"output-{index}{suffix}"
        if job["modality"] == "image":
            path.write_bytes(PNG)
        elif job["modality"] == "video":
            path.write_bytes(b"\x00\x00\x00\x18ftypisom" + b"\x00" * 24)
        else:
            with wave.open(str(path), "wb") as handle:
                handle.setnchannels(1)
                handle.setsampwidth(2)
                handle.setframerate(8000)
                handle.writeframes(struct.pack("<h", 0) * 80)
        outputs.append({"target": target, "source": str(path)})
    response = {"outputs": outputs, "provider_job_id": "fixture-local"}
    if isinstance(collecting, str) and collecting:
        response["provider_job_id"] = collecting
    json.dump(response, sys.stdout)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Prepare and execute explicitly confirmed short-drama media jobs.

A configured adapter receives one bounded JSON job on stdin and returns local
output files on stdout. The adapter is launched without a shell and only after
a confirmation bound to the exact job and current project inputs. Optional
provider adapters can ship with this skill, but credentials and adapter config
remain outside creator projects.
"""

from __future__ import annotations

import argparse
import contextlib
import hashlib
import json
import os
import re
import stat
import subprocess
import sys
import tempfile
import uuid
from collections.abc import Iterator, Mapping
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
from typing import Any, BinaryIO

MINIMUM_PYTHON = (3, 9)
if sys.version_info < MINIMUM_PYTHON:
    raise SystemExit(
        "short-drama-produce needs Python {}.{} or newer".format(*MINIMUM_PYTHON)
    )

PROJECT_FILE = "short-drama.json"
PRODUCTION_ROOT = Path(".short-drama/production")
JOB_SCHEMA = "1.0"
JOB_ID_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,79}")
MAX_JOB_BYTES = 256 * 1024
MAX_RUN_RECORD_BYTES = 256 * 1024
MAX_ADAPTER_RESPONSE_BYTES = 1024 * 1024
MAX_OUTPUT_BYTES = 512 * 1024 * 1024
MAX_INPUT_BYTES = 50 * 1024 * 1024
MAX_TOTAL_INPUT_BYTES = 200 * 1024 * 1024
MAX_TIMEOUT_SECONDS = 3600
PUBLIC_ERROR_CATEGORIES = {
    "authentication",
    "configuration",
    "contract",
    "invalid_request",
    "network",
    "permission",
    "provider_response",
    "rate_limit",
    "server",
    "timeout",
}
PUBLIC_ERROR_TOKEN_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,199}")
ALLOWED_JOB_KEYS = {
    "schema_version",
    "job_id",
    "modality",
    "adapter",
    "prompt",
    "source",
    "source_entry",
    "references",
    "reference_bindings",
    "outputs",
    "parameters",
    "overwrite",
}
STORED_EXECUTION_KEYS = ALLOWED_JOB_KEYS | {"inputs"}
STORED_JOB_KEYS = STORED_EXECUTION_KEYS | {"fingerprint", "prepared_at"}
LEGACY_STORED_EXECUTION_KEYS = STORED_EXECUTION_KEYS - {
    "source_entry",
    "reference_bindings",
}
LEGACY_STORED_JOB_KEYS = LEGACY_STORED_EXECUTION_KEYS | {
    "fingerprint",
    "prepared_at",
}
SECRET_KEYS = {
    "authorization",
    "credential",
    "credentials",
    "password",
    "secret",
    "token",
    "access_token",
    "api_key",
    "apikey",
}
MEDIA_EXTENSIONS = {
    "image": {".png", ".jpg", ".jpeg", ".webp"},
    "video": {".mp4", ".mov", ".webm"},
    "tts": {".wav", ".mp3", ".m4a", ".aac", ".flac", ".opus"},
    "music": {".wav", ".mp3", ".m4a", ".aac", ".flac", ".opus"},
}
MEDIA_TYPES = {
    ".png": "image/png",
    ".jpg": "image/jpeg",
    ".jpeg": "image/jpeg",
    ".webp": "image/webp",
    ".mp4": "video/mp4",
    ".mov": "video/quicktime",
    ".webm": "video/webm",
    ".wav": "audio/wav",
    ".mp3": "audio/mpeg",
    ".m4a": "audio/mp4",
    ".aac": "audio/aac",
    ".flac": "audio/flac",
    ".opus": "audio/ogg",
}
CREATOR_SOURCE_NAMES = {
    "图片提示词.md": "image",
    "视频提示词.md": "video",
    # A shot's frozen keyframe is a producible image and lives nowhere else, so
    # without this the start frame a video job wants to bind has no stage that
    # can render it.
    "分镜.md": "image",
}
# Which H2 IDs a creator source may select, and where that entry keeps its
# copyable body.
CREATOR_SOURCE_ENTRIES = {
    "图片提示词.md": ("IMG-", r"可复制(?:通用)?提示词", "参考"),
    "视频提示词.md": ("MOTION-", r"可复制(?:通用)?提示词", "输入参考图"),
    "分镜.md": ("SHOT-", r"冻结关键帧提示词", "输入参考图"),
}

REF_SLOT_RE = re.compile(r"REF-[A-Z0-9][A-Z0-9-]{0,79}")
# A slot head, not a bare substring: a real reference file may legitimately be
# named `plan-sheet.png`, and that is not a creator-supplied declaration.
PLAN_SLOT_HEAD_RE = re.compile(r"PLAN-[A-Za-z0-9][A-Za-z0-9-]{0,79}（顺序：")
PLAN_SLOT_RE = re.compile(
    r"PLAN-[A-Za-z0-9][A-Za-z0-9-]{0,79}（顺序：[1-9]\d*）· "
    r"(?:IMG|SHOT)-[A-Za-z0-9-]+《[^》\n]+》（[^）\n]*）"
)
SOURCE_ENTRY_RE = re.compile(r"[A-Z][A-Z0-9-]{1,99}")
REFERENCE_SUFFIX_RE = r"(?:png|jpe?g|webp)"
# 用途 is the creator-facing question "what does this picture decide here"; the
# job's own `role` is its production translation. The segment is optional so a
# declaration written before that field existed still prepares, and so the
# creator-side diagnostic stays with creator_markdown_check.py.
REFERENCE_LINE_RE = re.compile(
    rf"(REF-[A-Z0-9][A-Z0-9-]{{0,79}})（顺序：([1-9]\d*)）· "
    rf"([^；\n]+?\.{REFERENCE_SUFFIX_RE})《([^》\n]+)》"
    r"（(?:用途：[^；）\n]+；)?控制：([^；）\n]+)；不得控制：([^）\n]+)）",
    re.IGNORECASE,
)


class ConfirmationRequiredError(RuntimeError):
    """The exact current job has not been explicitly confirmed."""


class AdapterError(RuntimeError):
    """A configured media adapter failed or broke its output contract."""

    def __init__(
        self, message: str, *, public_error: Mapping[str, Any] | None = None
    ) -> None:
        super().__init__(message)
        self.public_error = dict(public_error) if public_error is not None else None


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")


def sha256_bytes(content: bytes) -> str:
    return hashlib.sha256(content).hexdigest()


def find_project(start: Path) -> Path:
    candidate = start.expanduser().resolve()
    if candidate.is_file():
        candidate = candidate.parent
    for directory in (candidate, *candidate.parents):
        if (directory / PROJECT_FILE).is_file():
            return directory
    raise FileNotFoundError(f"no {PROJECT_FILE} found from {start}")


def _relative_path(value: object, *, output: bool = False) -> str:
    if not isinstance(value, str):
        raise ValueError("project paths must be strings")
    raw = value.replace("\\", "/")
    pure = PurePosixPath(raw)
    if not raw or pure.is_absolute() or any(part in {"", ".", ".."} for part in pure.parts):
        raise ValueError(f"unsafe project-relative path: {value}")
    if pure.parts[0].casefold() == ".short-drama" or pure.name.casefold() == PROJECT_FILE:
        raise ValueError(f"operational project path is not allowed: {value}")
    if output:
        parts = pure.parts
        top_level_production = len(parts) >= 2 and parts[0].casefold() == "production"
        episode_production = (
            len(parts) >= 4
            and parts[0] in {"剧集", "episodes"}
            and re.fullmatch(r"EP\d{3,}", parts[1], re.IGNORECASE) is not None
            and parts[2] in {"制作成果", "production"}
        )
        if not top_level_production and not episode_production:
            raise ValueError(
                "media outputs must use top-level production/ or "
                "剧集|episodes/<EP>/制作成果|production/"
            )
    return pure.as_posix()


def _project_file(root: Path, relative: str, *, create_parent: bool = False) -> Path:
    target = root / relative
    current = root
    for part in PurePosixPath(relative).parts[:-1]:
        current /= part
        if current.exists() and (current.is_symlink() or not current.is_dir()):
            raise ValueError(f"unsafe project directory: {part}")
        if create_parent and not current.exists():
            current.mkdir()
    if target.exists() and (target.is_symlink() or not target.is_file()):
        raise ValueError(f"unsafe project file: {relative}")
    if not target.parent.resolve().is_relative_to(root):
        raise ValueError(f"path escapes project root: {relative}")
    return target


def _is_link_or_reparse(details: os.stat_result) -> bool:
    attributes = getattr(details, "st_file_attributes", 0)
    reparse_flag = getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0)
    return stat.S_ISLNK(details.st_mode) or bool(attributes & reparse_flag)


@contextlib.contextmanager
def _open_project_input(root: Path, relative: str) -> Iterator[BinaryIO]:
    """Open one regular project input without following path components on POSIX."""
    parts = PurePosixPath(relative).parts
    if not parts:
        raise ValueError("job input path is empty")
    if os.name != "nt" and os.open in os.supports_dir_fd:
        directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
        nofollow = getattr(os, "O_NOFOLLOW", 0)
        directory_fd = os.open(root, directory_flags)
        file_fd: int | None = None
        try:
            for part in parts[:-1]:
                next_fd = os.open(
                    part,
                    directory_flags | nofollow,
                    dir_fd=directory_fd,
                )
                os.close(directory_fd)
                directory_fd = next_fd
            file_fd = os.open(
                parts[-1], os.O_RDONLY | nofollow, dir_fd=directory_fd
            )
            details = os.fstat(file_fd)
            if not stat.S_ISREG(details.st_mode):
                raise ValueError(f"job input is not a regular file: {relative}")
            with os.fdopen(file_fd, "rb", closefd=True) as handle:
                file_fd = None
                yield handle
        except FileNotFoundError as exc:
            raise FileNotFoundError(f"job input is missing: {relative}") from exc
        except OSError as exc:
            raise ValueError(f"unsafe job input path: {relative}") from exc
        finally:
            if file_fd is not None:
                os.close(file_fd)
            os.close(directory_fd)
        return

    # Windows lacks portable openat/O_NOFOLLOW support. Reject reparse/symlink
    # components, pin the final file handle, and verify its identity before use.
    path = root
    for part in parts:
        path /= part
        try:
            details = path.lstat()
        except FileNotFoundError as exc:
            raise FileNotFoundError(f"job input is missing: {relative}") from exc
        if _is_link_or_reparse(details):
            raise ValueError(f"unsafe job input path: {relative}")
    if not path.resolve().is_relative_to(root):
        raise ValueError(f"job input escapes project root: {relative}")
    before = path.stat(follow_symlinks=False)
    if not stat.S_ISREG(before.st_mode):
        raise ValueError(f"job input is not a regular file: {relative}")
    with path.open("rb") as handle:
        opened = os.fstat(handle.fileno())
        if (opened.st_dev, opened.st_ino) != (before.st_dev, before.st_ino):
            raise ValueError(f"job input changed while opening: {relative}")
        yield handle


def _hash_project_file(root: Path, relative: str) -> str:
    digest = hashlib.sha256()
    size = 0
    with _open_project_input(root, relative) as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            size += len(chunk)
            if size > MAX_INPUT_BYTES:
                raise ValueError(f"job input exceeds the size limit: {relative}")
            digest.update(chunk)
    return digest.hexdigest()


def _snapshot_inputs(
    root: Path, job: Mapping[str, Any], snapshot_root: Path
) -> None:
    inputs = job.get("inputs")
    if not isinstance(inputs, Mapping):
        raise ConfirmationRequiredError("stored job inputs are invalid")
    total = 0
    for relative_value, expected_value in inputs.items():
        relative = _relative_path(relative_value)
        if not isinstance(expected_value, str) or re.fullmatch(
            r"[0-9a-f]{64}", expected_value
        ) is None:
            raise ConfirmationRequiredError("stored job input hash is invalid")
        target = snapshot_root.joinpath(*PurePosixPath(relative).parts)
        target.parent.mkdir(parents=True, exist_ok=True)
        digest = hashlib.sha256()
        size = 0
        try:
            with _open_project_input(root, relative) as incoming, target.open(
                "xb"
            ) as outgoing:
                for chunk in iter(lambda: incoming.read(1024 * 1024), b""):
                    size += len(chunk)
                    total += len(chunk)
                    if size > MAX_INPUT_BYTES or total > MAX_TOTAL_INPUT_BYTES:
                        raise ConfirmationRequiredError(
                            "job inputs exceed the production size limit"
                        )
                    digest.update(chunk)
                    outgoing.write(chunk)
        except (FileNotFoundError, OSError, ValueError) as exc:
            raise ConfirmationRequiredError(
                "job inputs changed; prepare and confirm again"
            ) from exc
        if digest.hexdigest() != expected_value:
            raise ConfirmationRequiredError(
                "job inputs changed; prepare and confirm again"
            )


def _canonical(document: object) -> bytes:
    return json.dumps(
        document, ensure_ascii=False, sort_keys=True, separators=(",", ":")
    ).encode("utf-8")


def _check_metadata_parts(parts: tuple[str, ...]) -> None:
    if any(
        not part
        or part in {".", ".."}
        or "/" in part
        or "\\" in part
        for part in parts
    ):
        raise ValueError("production metadata path is invalid")


@contextlib.contextmanager
def _metadata_directory(
    root: Path, parts: tuple[str, ...], *, create: bool
) -> Iterator[tuple[Path, int | None]]:
    """Pin one metadata directory without following project-controlled parents."""
    root = root.resolve()
    _check_metadata_parts(parts)
    all_parts = (".short-drama", "production", *parts)
    directory = root.joinpath(*all_parts)
    if os.name != "nt" and os.open in os.supports_dir_fd:
        flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
        nofollow = getattr(os, "O_NOFOLLOW", 0)
        directory_fd = os.open(root, flags)
        try:
            try:
                for part in all_parts:
                    try:
                        next_fd = os.open(
                            part, flags | nofollow, dir_fd=directory_fd
                        )
                    except FileNotFoundError:
                        if not create:
                            raise
                        try:
                            os.mkdir(part, mode=0o700, dir_fd=directory_fd)
                        except FileExistsError:
                            pass
                        next_fd = os.open(
                            part, flags | nofollow, dir_fd=directory_fd
                        )
                    details = os.fstat(next_fd)
                    if not stat.S_ISDIR(details.st_mode):
                        os.close(next_fd)
                        raise ValueError("production metadata directory is unsafe")
                    os.close(directory_fd)
                    directory_fd = next_fd
            except FileNotFoundError:
                raise
            except OSError as exc:
                raise ValueError("production metadata directory is unsafe") from exc
            yield directory, directory_fd
        finally:
            os.close(directory_fd)
        return

    current = root
    for part in all_parts:
        current /= part
        try:
            details = current.lstat()
        except FileNotFoundError:
            if not create:
                raise
            try:
                current.mkdir(mode=0o700)
            except FileExistsError:
                pass
            details = current.lstat()
        if _is_link_or_reparse(details) or not stat.S_ISDIR(details.st_mode):
            raise ValueError("production metadata directory is unsafe")
    if not current.resolve().is_relative_to(root):
        raise ValueError("production metadata directory escapes the project")
    yield current, None


def _metadata_atomic_json(
    root: Path, directory_parts: tuple[str, ...], name: str, document: Mapping[str, Any]
) -> None:
    _check_metadata_parts((name,))
    content = _canonical(document) + b"\n"
    temporary_name = f".{name}.{uuid.uuid4().hex}.tmp"
    with _metadata_directory(root, directory_parts, create=True) as (
        directory,
        directory_fd,
    ):
        if directory_fd is not None:
            descriptor = -1
            try:
                descriptor = os.open(
                    temporary_name,
                    os.O_WRONLY
                    | os.O_CREAT
                    | os.O_EXCL
                    | getattr(os, "O_NOFOLLOW", 0),
                    0o600,
                    dir_fd=directory_fd,
                )
                with os.fdopen(descriptor, "wb", closefd=True) as handle:
                    descriptor = -1
                    handle.write(content)
                    handle.flush()
                    os.fsync(handle.fileno())
                os.replace(
                    temporary_name,
                    name,
                    src_dir_fd=directory_fd,
                    dst_dir_fd=directory_fd,
                )
                os.fsync(directory_fd)
            finally:
                if descriptor >= 0:
                    os.close(descriptor)
                try:
                    os.unlink(temporary_name, dir_fd=directory_fd)
                except FileNotFoundError:
                    pass
            return

        before = directory.stat(follow_symlinks=False)
        temporary = directory / temporary_name
        try:
            with temporary.open("xb") as handle:
                opened = os.fstat(handle.fileno())
                if not stat.S_ISREG(opened.st_mode):
                    raise ValueError("production metadata file is unsafe")
                handle.write(content)
                handle.flush()
                os.fsync(handle.fileno())
            after = directory.lstat()
            if (
                _is_link_or_reparse(after)
                or not stat.S_ISDIR(after.st_mode)
                or (after.st_dev, after.st_ino) != (before.st_dev, before.st_ino)
            ):
                raise ValueError("production metadata directory changed")
            os.replace(temporary, directory / name)
        finally:
            try:
                temporary.unlink()
            except FileNotFoundError:
                pass


def _metadata_read_json(
    root: Path, directory_parts: tuple[str, ...], name: str, *, maximum: int
) -> object:
    _check_metadata_parts((name,))
    with _metadata_directory(root, directory_parts, create=False) as (
        directory,
        directory_fd,
    ):
        if directory_fd is not None:
            descriptor = os.open(
                name,
                os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0),
                dir_fd=directory_fd,
            )
            with os.fdopen(descriptor, "rb", closefd=True) as handle:
                details = os.fstat(handle.fileno())
                if not stat.S_ISREG(details.st_mode) or details.st_size > maximum:
                    raise ValueError("production metadata file is unsafe")
                raw = handle.read(maximum + 1)
        else:
            path = directory / name
            before = path.lstat()
            if (
                _is_link_or_reparse(before)
                or not stat.S_ISREG(before.st_mode)
                or before.st_size > maximum
            ):
                raise ValueError("production metadata file is unsafe")
            with path.open("rb") as handle:
                opened = os.fstat(handle.fileno())
                if (opened.st_dev, opened.st_ino) != (before.st_dev, before.st_ino):
                    raise ValueError("production metadata file changed while opening")
                raw = handle.read(maximum + 1)
        if len(raw) > maximum:
            raise ValueError("production metadata file is too large")
        return json.loads(raw.decode("utf-8"))


def _metadata_unlink(
    root: Path, directory_parts: tuple[str, ...], name: str
) -> None:
    _check_metadata_parts((name,))
    with _metadata_directory(root, directory_parts, create=False) as (
        directory,
        directory_fd,
    ):
        if directory_fd is not None:
            os.unlink(name, dir_fd=directory_fd)
        else:
            path = directory / name
            details = path.lstat()
            if _is_link_or_reparse(details) or not stat.S_ISREG(details.st_mode):
                raise ValueError("production metadata file is unsafe")
            path.unlink()


def _metadata_json_names(root: Path, directory_parts: tuple[str, ...]) -> list[str]:
    try:
        with _metadata_directory(root, directory_parts, create=False) as (
            directory,
            directory_fd,
        ):
            names = os.listdir(directory_fd if directory_fd is not None else directory)
    except FileNotFoundError:
        return []
    return sorted(
        name
        for name in names
        if isinstance(name, str)
        and name.endswith(".json")
        and "/" not in name
        and "\\" not in name
    )


@contextlib.contextmanager
def _project_lock(root: Path) -> Iterator[None]:
    with _metadata_directory(root, (), create=True) as (directory, directory_fd):
        if directory_fd is not None:
            descriptor = os.open(
                "lock",
                os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0),
                0o600,
                dir_fd=directory_fd,
            )
            details = os.fstat(descriptor)
            if not stat.S_ISREG(details.st_mode):
                os.close(descriptor)
                raise ValueError("production lock is unsafe")
            handle_context = os.fdopen(descriptor, "a+b", closefd=True)
        else:
            lock_path = directory / "lock"
            before: os.stat_result | None
            try:
                before = lock_path.lstat()
            except FileNotFoundError:
                before = None
            else:
                if _is_link_or_reparse(before) or not stat.S_ISREG(before.st_mode):
                    raise ValueError("production lock is unsafe")
            handle_context = lock_path.open("a+b")
            opened = os.fstat(handle_context.fileno())
            after = lock_path.lstat()
            if (
                not stat.S_ISREG(opened.st_mode)
                or _is_link_or_reparse(after)
                or not stat.S_ISREG(after.st_mode)
                or (opened.st_dev, opened.st_ino) != (after.st_dev, after.st_ino)
                or (
                    before is not None
                    and (before.st_dev, before.st_ino)
                    != (opened.st_dev, opened.st_ino)
                )
            ):
                handle_context.close()
                raise ValueError("production lock changed while opening")
        with handle_context as handle:
            if os.name == "nt":
                import msvcrt

                handle.seek(0, os.SEEK_END)
                if handle.tell() == 0:
                    handle.write(b"0")
                    handle.flush()
                handle.seek(0)
                locking = getattr(msvcrt, "locking")
                lock = getattr(msvcrt, "LK_LOCK")
                unlock = getattr(msvcrt, "LK_UNLCK")
                locking(handle.fileno(), lock, 1)
                try:
                    yield
                finally:
                    handle.seek(0)
                    locking(handle.fileno(), unlock, 1)
            else:
                import fcntl

                fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
                try:
                    yield
                finally:
                    fcntl.flock(handle.fileno(), fcntl.LOCK_UN)


def _job_key(job_id: str) -> str:
    return sha256_bytes(job_id.encode("utf-8"))[:24]


def _job_path(root: Path, job_id: str) -> Path:
    return root / PRODUCTION_ROOT / "jobs" / f"{_job_key(job_id)}.json"


def _confirmation_path(root: Path, job_id: str) -> Path:
    return root / PRODUCTION_ROOT / "confirmations" / f"{_job_key(job_id)}.json"


def _run_directory(root: Path, job_id: str) -> Path:
    return root / PRODUCTION_ROOT / "runs" / _job_key(job_id)


def _contains_secret_key(value: object) -> bool:
    if isinstance(value, Mapping):
        return any(
            str(key).casefold() in SECRET_KEYS or _contains_secret_key(child)
            for key, child in value.items()
        )
    if isinstance(value, list):
        return any(_contains_secret_key(child) for child in value)
    return False


def _string_list(value: object, *, label: str, limit: int = 32) -> list[str]:
    if not isinstance(value, list) or not all(isinstance(item, str) for item in value):
        raise ValueError(f"{label} must be a string list")
    if len(value) > limit:
        raise ValueError(f"{label} has too many entries")
    return list(value)


def _scope_list(value: object, *, label: str) -> list[str]:
    items = _string_list(value, label=label, limit=32)
    if not items or any(not item.strip() or len(item) > 200 for item in items):
        raise ValueError(f"{label} must contain non-empty bounded text")
    normalized = [item.strip() for item in items]
    folded = [item.casefold() for item in normalized]
    if len(folded) != len(set(folded)):
        raise ValueError(f"{label} must not contain duplicates")
    return normalized


def _canonical_creator_source_modality(source: str | None) -> str | None:
    if source is None:
        return None
    path = PurePosixPath(source)
    if (
        len(path.parts) != 3
        or path.parts[0] not in {"剧集", "episodes"}
        or not path.parts[1]
    ):
        return None
    return CREATOR_SOURCE_NAMES.get(path.name)


def _normalize_reference_bindings(value: object) -> list[dict[str, Any]]:
    if value is None:
        return []
    if not isinstance(value, list) or len(value) > 16:
        raise ValueError("reference_bindings must be a list of at most sixteen entries")
    normalized: list[dict[str, Any]] = []
    for index, binding in enumerate(value, 1):
        label = f"reference_bindings[{index}]"
        if not isinstance(binding, Mapping) or set(binding) != {
            "slot_id",
            "order",
            "path",
            "label",
            "role",
            "may_control",
            "must_not_control",
        }:
            raise ValueError(f"{label} fields are invalid")
        slot_id = binding.get("slot_id")
        order = binding.get("order")
        raw_label = binding.get("label")
        role = binding.get("role")
        if not isinstance(slot_id, str) or REF_SLOT_RE.fullmatch(slot_id) is None:
            raise ValueError(f"{label} slot_id is invalid")
        if not isinstance(order, int) or isinstance(order, bool) or order < 1:
            raise ValueError(f"{label} order is invalid")
        if (
            not isinstance(raw_label, str)
            or not raw_label.strip()
            or len(raw_label) > 200
            or re.search(r"[\u3400-\u9fff]", raw_label) is None
        ):
            raise ValueError(f"{label} label must contain Chinese text")
        if not isinstance(role, str) or not role.strip() or len(role) > 80:
            raise ValueError(f"{label} role is invalid")
        may_control = _scope_list(
            binding.get("may_control"), label=f"{label}.may_control"
        )
        must_not_control = _scope_list(
            binding.get("must_not_control"), label=f"{label}.must_not_control"
        )
        if {item.casefold() for item in may_control} & {
            item.casefold() for item in must_not_control
        }:
            raise ValueError(f"{label} control scopes must not overlap")
        normalized.append(
            {
                "slot_id": slot_id,
                "order": order,
                "path": _relative_path(binding.get("path")),
                "label": raw_label.strip(),
                "role": role.strip(),
                "may_control": may_control,
                "must_not_control": must_not_control,
            }
        )
    normalized.sort(key=lambda item: int(item["order"]))
    slots = [str(item["slot_id"]) for item in normalized]
    orders = [int(item["order"]) for item in normalized]
    paths = [str(item["path"]) for item in normalized]
    if len(slots) != len(set(slots)):
        raise ValueError("reference binding slot_ids must be unique")
    if orders != list(range(1, len(orders) + 1)):
        raise ValueError("reference binding order must be unique and contiguous from 1")
    if len(paths) != len(set(paths)):
        raise ValueError("reference binding paths must be unique")
    return normalized


def _markdown_section(document: str, source_entry: str) -> str:
    heading = re.compile(
        rf"^##\s+`?{re.escape(source_entry)}`?(?:\s+·.*)?\s*$", re.MULTILINE
    )
    matches = list(heading.finditer(document))
    if not matches:
        raise ValueError(f"source entry is missing from Markdown: {source_entry}")
    if len(matches) != 1:
        raise ValueError(f"source entry is duplicated in Markdown: {source_entry}")
    match = matches[0]
    next_heading = re.search(r"^##\s+", document[match.end() :], re.MULTILINE)
    end = match.end() + next_heading.start() if next_heading is not None else len(document)
    return document[match.start() : end]


def _copyable_prompt(
    section: str, heading: str = r"可复制(?:通用)?提示词"
) -> str:
    markers = list(
        re.finditer(rf"^###\s+{heading}\s*$", section, re.MULTILINE)
    )
    if not markers:
        raise ValueError("source entry has no copyable prompt")
    if len(markers) != 1:
        raise ValueError("source entry has duplicate copyable prompts")
    marker = markers[0]
    body = section[marker.end() :]
    following = re.search(r"^###\s+|^##\s+", body, re.MULTILINE)
    if following is not None:
        body = body[: following.start()]
    body_lines = [line for line in body.splitlines() if line.strip()]
    if any(not line.startswith(">") for line in body_lines):
        raise ValueError("source entry copyable prompt contains unquoted content")
    lines = [line[1:].lstrip() for line in body_lines]
    if not lines or not "\n".join(lines).strip():
        raise ValueError("source entry copyable prompt is empty")
    return "\n".join(lines).strip()


def _scope_items(value: str) -> list[str]:
    return [item.strip() for item in re.split(r"[、,，]", value) if item.strip()]


def _contains_ref_token(value: str) -> bool:
    return "ref-" in value.casefold()


def _contains_plan_token(value: str) -> bool:
    return "plan-" in value.casefold()


def _markdown_reference_bindings(
    section: str, *, field_name: str, creator_supplied_ok: bool = False
) -> list[dict[str, Any]]:
    lines = re.findall(
        rf"^- {re.escape(field_name)}：(.+)$", section, re.MULTILINE
    )
    if not lines:
        raise ValueError(f"source entry has no {field_name} declaration")
    if len(lines) != 1:
        raise ValueError(f"source entry has duplicate {field_name} declarations")
    value = lines[0].strip()
    if (
        field_name == "输入参考图"
        and not _contains_ref_token(value)
        and re.fullmatch(r"无(?:（[^）\n]+）)?。?", value)
    ):
        return []
    if (
        field_name == "参考"
        and not _contains_ref_token(value)
        and re.fullmatch(r"无(?:外部参考)?(?:；[^\n]*)?。?", value)
    ):
        return []
    # A `PLAN-...` slot is a picture the creator attaches in their own tool, so
    # this suite has no file to send. The rendering job that draws a shot's own
    # start frame is unaffected -- it works from the frozen keyframe text, the
    # way it already does for 「待补参考图」 -- so only the job that would have
    # sent these pictures as model inputs refuses. Matched as a slot head rather
    # than a bare substring, or a real file named `plan-sheet.png` would refuse.
    if PLAN_SLOT_HEAD_RE.search(value):
        if not creator_supplied_ok:
            raise ValueError(
                "source entry declares creator-supplied references (PLAN-...); "
                "bind the real files as REF-... before producing"
            )
        value = PLAN_SLOT_RE.sub("", value).strip("；;。 ")
        if not value:
            return []
    matches = list(REFERENCE_LINE_RE.finditer(value))
    if not matches:
        raise ValueError("source entry input-reference declaration is invalid")
    cursor = 0
    for index, match in enumerate(matches):
        separator = value[cursor : match.start()]
        if separator != ("" if index == 0 else "；"):
            raise ValueError("source entry input-reference declaration is invalid")
        cursor = match.end()
    if value[cursor:] not in {"", "。"}:
        raise ValueError("source entry input-reference declaration is invalid")
    return [
        {
            "slot_id": match.group(1),
            "order": int(match.group(2)),
            "path": _relative_path(match.group(3)),
            "label": match.group(4).strip(),
            "may_control": _scope_items(match.group(5)),
            "must_not_control": _scope_items(match.group(6)),
        }
        for match in matches
    ]


def _verify_markdown_source(
    root: Path,
    *,
    source: str,
    source_entry: str,
    prompt: str,
    bindings: list[dict[str, Any]],
) -> None:
    source_path = _project_file(root, source)
    if source_path.suffix.casefold() != ".md":
        raise ValueError("source_entry requires a Markdown source")
    document = source_path.read_text(encoding="utf-8")
    section = _markdown_section(document, source_entry)
    _, heading, field_name = CREATOR_SOURCE_ENTRIES[source_path.name]
    if _copyable_prompt(section, heading) != prompt.strip():
        raise ValueError("job prompt does not match the selected source entry")
    declared = _markdown_reference_bindings(
        section,
        field_name=field_name,
        # 分镜.md is the entry that renders a shot's own start frame. Its
        # reference field records creative intent, not this job's inputs.
        creator_supplied_ok=source_path.name == "分镜.md",
    )
    comparable = [
        {key: binding[key] for key in (
            "slot_id",
            "order",
            "path",
            "label",
            "may_control",
            "must_not_control",
        )}
        for binding in bindings
    ]
    if declared != comparable:
        raise ValueError("job reference bindings do not match the selected source entry")


def _normalize_job(root: Path, raw: object) -> dict[str, Any]:
    if not isinstance(raw, Mapping) or set(raw) - ALLOWED_JOB_KEYS:
        raise ValueError("job contains unsupported fields")
    if raw.get("schema_version", JOB_SCHEMA) != JOB_SCHEMA:
        raise ValueError("unsupported job schema")
    job_id = raw.get("job_id")
    if not isinstance(job_id, str) or JOB_ID_RE.fullmatch(job_id) is None:
        raise ValueError("job_id must be a portable 1-80 character identifier")
    modality = raw.get("modality")
    if modality not in MEDIA_EXTENSIONS:
        raise ValueError("modality must be image, video, tts, or music")
    adapter = raw.get("adapter")
    if not isinstance(adapter, str) or JOB_ID_RE.fullmatch(adapter) is None:
        raise ValueError("adapter must be a portable profile name")
    prompt = raw.get("prompt")
    if not isinstance(prompt, str) or not prompt.strip() or len(prompt) > 100_000:
        raise ValueError("prompt must be non-empty and at most 100000 characters")
    source_raw = raw.get("source")
    source = _relative_path(source_raw) if source_raw is not None else None
    source_entry = raw.get("source_entry")
    if source_entry is not None and (
        not isinstance(source_entry, str)
        or SOURCE_ENTRY_RE.fullmatch(source_entry) is None
    ):
        raise ValueError("source_entry must be a visible uppercase Markdown entry ID")
    if source_entry is not None and source is None:
        raise ValueError("source_entry requires source")
    creator_source_modality = _canonical_creator_source_modality(source)
    if creator_source_modality == modality and source_entry is None:
        raise ValueError("creator Markdown source requires source_entry")
    if creator_source_modality is not None and modality != creator_source_modality and not (
        creator_source_modality == "video"
        and modality == "music"
        and source_entry is None
    ):
        raise ValueError("creator Markdown source does not match the job modality")
    expected_entry_prefix = (
        CREATOR_SOURCE_ENTRIES[PurePosixPath(source).name][0]
        if source is not None
        and PurePosixPath(source).name in CREATOR_SOURCE_ENTRIES
        and creator_source_modality == modality
        else None
    )
    if source_entry is not None and creator_source_modality != modality:
        raise ValueError("source_entry requires the canonical creator Markdown path")
    if source_entry is not None and (
        expected_entry_prefix is None
        or not source_entry.startswith(expected_entry_prefix)
    ):
        raise ValueError("source_entry does not match the job modality")
    references_supplied = "references" in raw
    supplied_references = [
        _relative_path(path)
        for path in _string_list(raw.get("references", []), label="references", limit=16)
    ]
    reference_bindings = _normalize_reference_bindings(raw.get("reference_bindings"))
    binding_references = [str(binding["path"]) for binding in reference_bindings]
    if (
        references_supplied
        and (reference_bindings or source_entry is not None)
        and supplied_references != binding_references
    ):
        raise ValueError("references must match reference_bindings order")
    references = binding_references if reference_bindings else supplied_references
    if source_entry is not None:
        _verify_markdown_source(
            root,
            source=str(source),
            source_entry=source_entry,
            prompt=prompt,
            bindings=reference_bindings,
        )
    outputs = [
        _relative_path(path, output=True)
        for path in _string_list(raw.get("outputs"), label="outputs", limit=16)
    ]
    if not outputs or len(set(outputs)) != len(outputs):
        raise ValueError("outputs must contain unique target paths")
    for output_path in outputs:
        if PurePosixPath(output_path).suffix.casefold() not in MEDIA_EXTENSIONS[str(modality)]:
            raise ValueError(f"output extension does not match {modality}: {output_path}")
        _project_file(root, output_path)
    parameters = raw.get("parameters", {})
    if not isinstance(parameters, Mapping):
        raise ValueError("parameters must be an object")
    if _contains_secret_key(parameters):
        raise ValueError("job parameters must not contain credentials or secrets")
    if len(_canonical(parameters)) > 64 * 1024:
        raise ValueError("job parameters are too large")
    input_paths = ([source] if source is not None else []) + references
    if len(set(input_paths)) != len(input_paths):
        raise ValueError("source and references must be unique")
    input_hashes = {path: _hash_project_file(root, path) for path in input_paths}
    execution = {
        "schema_version": JOB_SCHEMA,
        "job_id": job_id,
        "modality": modality,
        "adapter": adapter,
        "prompt": prompt,
        "source": source,
        "source_entry": source_entry,
        "references": references,
        "reference_bindings": reference_bindings,
        "outputs": outputs,
        "parameters": dict(parameters),
        "overwrite": raw.get("overwrite", False),
        "inputs": input_hashes,
    }
    if not isinstance(execution["overwrite"], bool):
        raise ValueError("overwrite must be a boolean")
    execution["fingerprint"] = sha256_bytes(_canonical(execution))
    execution["prepared_at"] = utc_now()
    return execution


def prepare_job(root: Path, job_file: Path) -> dict[str, Any]:
    root = find_project(root)
    if job_file.stat().st_size > MAX_JOB_BYTES:
        raise ValueError("job file is too large")
    raw = json.loads(job_file.read_text(encoding="utf-8"))
    job = _normalize_job(root, raw)
    with _project_lock(root):
        if _active_run(root, str(job["job_id"])) is not None:
            raise RuntimeError("this job is already running")
        job_name = f"{_job_key(str(job['job_id']))}.json"
        _metadata_atomic_json(root, ("jobs",), job_name, job)
        try:
            _metadata_unlink(root, ("confirmations",), job_name)
        except FileNotFoundError:
            pass
    return _preview(job)


def _validate_stored_job(
    root: Path, document: object, *, expected_job_id: str | None = None
) -> dict[str, Any]:
    root = root.resolve()
    if not isinstance(document, dict):
        raise ValueError("stored job fields are invalid")
    stored_keys = set(document)
    legacy = stored_keys == LEGACY_STORED_JOB_KEYS
    if not legacy and stored_keys != STORED_JOB_KEYS:
        raise ValueError("stored job fields are invalid")
    original = document
    document = dict(document)
    if legacy:
        document["source_entry"] = None
        document["reference_bindings"] = []
    job_id = document.get("job_id")
    if (
        not isinstance(job_id, str)
        or JOB_ID_RE.fullmatch(job_id) is None
        or (expected_job_id is not None and job_id != expected_job_id)
    ):
        raise ValueError("stored job id is invalid")
    if document.get("schema_version") != JOB_SCHEMA:
        raise ValueError("stored job schema is invalid")
    modality = document.get("modality")
    if modality not in MEDIA_EXTENSIONS:
        raise ValueError("stored job modality is invalid")
    adapter = document.get("adapter")
    if not isinstance(adapter, str) or JOB_ID_RE.fullmatch(adapter) is None:
        raise ValueError("stored job adapter is invalid")
    prompt = document.get("prompt")
    if not isinstance(prompt, str) or not prompt.strip() or len(prompt) > 100_000:
        raise ValueError("stored job prompt is invalid")
    source = document.get("source")
    if source is not None and _relative_path(source) != source:
        raise ValueError("stored job source is invalid")
    source_entry = document.get("source_entry")
    if source_entry is not None and (
        not isinstance(source_entry, str)
        or SOURCE_ENTRY_RE.fullmatch(source_entry) is None
        or source is None
    ):
        raise ValueError("stored job source entry is invalid")
    expected_entry_prefix = {"image": "IMG-", "video": "MOTION-"}.get(
        str(modality)
    )
    if source_entry is not None and (
        expected_entry_prefix is None
        or not source_entry.startswith(expected_entry_prefix)
    ):
        raise ValueError("stored job source entry has the wrong modality")
    creator_source_modality = _canonical_creator_source_modality(
        str(source) if source is not None else None
    )
    if not legacy and creator_source_modality == modality and source_entry is None:
        raise ValueError("stored creator Markdown job has no source entry")
    if creator_source_modality is not None and modality != creator_source_modality and not (
        creator_source_modality == "video"
        and modality == "music"
        and source_entry is None
    ):
        raise ValueError("stored creator Markdown source has the wrong modality")
    if source_entry is not None and creator_source_modality != modality:
        raise ValueError("stored job creator source path is invalid")
    references = _string_list(
        document.get("references"), label="stored references", limit=16
    )
    if any(_relative_path(reference) != reference for reference in references):
        raise ValueError("stored job references are invalid")
    reference_bindings = _normalize_reference_bindings(
        document.get("reference_bindings")
    )
    binding_references = [binding["path"] for binding in reference_bindings]
    if (
        source_entry is not None or reference_bindings
    ) and binding_references != references:
        raise ValueError("stored job references do not match reference bindings")
    outputs = _string_list(document.get("outputs"), label="stored outputs", limit=16)
    if not outputs or len(outputs) != len(set(outputs)):
        raise ValueError("stored job outputs are invalid")
    for output in outputs:
        if _relative_path(output, output=True) != output:
            raise ValueError("stored job output is invalid")
        if PurePosixPath(output).suffix.casefold() not in MEDIA_EXTENSIONS[str(modality)]:
            raise ValueError("stored job output extension is invalid")
        _project_file(root, output)
    parameters = document.get("parameters")
    if (
        not isinstance(parameters, Mapping)
        or _contains_secret_key(parameters)
        or len(_canonical(parameters)) > 64 * 1024
    ):
        raise ValueError("stored job parameters are invalid")
    if not isinstance(document.get("overwrite"), bool):
        raise ValueError("stored job overwrite flag is invalid")
    input_paths = ([source] if source is not None else []) + references
    if len(input_paths) != len(set(input_paths)):
        raise ValueError("stored job inputs are duplicated")
    inputs = document.get("inputs")
    if not isinstance(inputs, Mapping) or set(inputs) != set(input_paths):
        raise ValueError("stored job input hashes are invalid")
    if any(
        not isinstance(digest, str) or re.fullmatch(r"[0-9a-f]{64}", digest) is None
        for digest in inputs.values()
    ):
        raise ValueError("stored job input hash is invalid")
    fingerprint = document.get("fingerprint")
    execution_keys = (
        LEGACY_STORED_EXECUTION_KEYS if legacy else STORED_EXECUTION_KEYS
    )
    execution = {key: original[key] for key in execution_keys}
    if (
        not isinstance(fingerprint, str)
        or re.fullmatch(r"[0-9a-f]{64}", fingerprint) is None
        or sha256_bytes(_canonical(execution)) != fingerprint
    ):
        raise ValueError("stored job fingerprint is invalid")
    _parse_run_timestamp(document.get("prepared_at"), label="prepared_at")
    return document


def _read_job(root: Path, job_id: str) -> dict[str, Any]:
    if JOB_ID_RE.fullmatch(job_id) is None:
        raise ValueError("invalid job_id")
    document = _metadata_read_json(
        root,
        ("jobs",),
        f"{_job_key(job_id)}.json",
        maximum=MAX_JOB_BYTES,
    )
    return _validate_stored_job(root, document, expected_job_id=job_id)


def _preview(job: Mapping[str, Any]) -> dict[str, Any]:
    confirmation = f"CONFIRM {job['job_id']} {str(job['fingerprint'])[:12]}"
    return {
        "job_id": job["job_id"],
        "modality": job["modality"],
        "adapter": job["adapter"],
        "count": len(job["outputs"]),
        "prompt": job["prompt"],
        "source": job["source"],
        "source_entry": job["source_entry"],
        "references": job["references"],
        "reference_bindings": job["reference_bindings"],
        "outputs": job["outputs"],
        "parameters": job["parameters"],
        "overwrite": job["overwrite"],
        "confirmation": confirmation,
        "state": "needs_confirmation",
    }


def confirm_job(root: Path, *, job_id: str, confirmation: str) -> dict[str, Any]:
    root = find_project(root)
    with _project_lock(root):
        if _active_run(root, job_id) is not None:
            raise RuntimeError("this job is already running")
        job = _read_job(root, job_id)
        expected = _preview(job)["confirmation"]
        if confirmation != expected:
            raise ConfirmationRequiredError("confirmation does not match the exact current job")
        receipt = {
            "schema_version": JOB_SCHEMA,
            "job_id": job_id,
            "fingerprint": job["fingerprint"],
            "confirmed_at": utc_now(),
            "consumed_at": None,
            "run_id": None,
        }
        _metadata_atomic_json(
            root, ("confirmations",), f"{_job_key(job_id)}.json", receipt
        )
    return {"job_id": job_id, "state": "confirmed"}


def _inputs_current(root: Path, job: Mapping[str, Any]) -> bool:
    inputs = job.get("inputs")
    if not isinstance(inputs, Mapping):
        return False
    try:
        return all(_hash_project_file(root, str(path)) == digest for path, digest in inputs.items())
    except (FileNotFoundError, OSError, ValueError):
        return False


def _load_adapter(config_path: Path, profile: str, root: Path) -> tuple[list[str], int]:
    resolved = config_path.expanduser().resolve()
    if resolved.is_relative_to(root):
        raise ValueError("adapter config must live outside the project")
    document = json.loads(resolved.read_text(encoding="utf-8"))
    adapters = document.get("adapters") if isinstance(document, Mapping) else None
    selected = adapters.get(profile) if isinstance(adapters, Mapping) else None
    if not isinstance(selected, Mapping) or set(selected) - {"command", "timeout_seconds"}:
        raise ValueError(f"adapter profile is missing or invalid: {profile}")
    command = _string_list(selected.get("command"), label="adapter command", limit=32)
    if not command or any(not part for part in command):
        raise ValueError("adapter command must be a non-empty argv list")
    timeout = selected.get("timeout_seconds", 300)
    if not isinstance(timeout, int) or isinstance(timeout, bool) or not 1 <= timeout <= MAX_TIMEOUT_SECONDS:
        raise ValueError(f"adapter timeout must be 1-{MAX_TIMEOUT_SECONDS} seconds")
    return command, timeout


def _generic_adapter_error(
    profile: str,
    *,
    category: str,
    code: str,
    retryable: bool,
) -> dict[str, Any]:
    return {
        "provider": profile,
        "category": category,
        "code": code,
        "retryable": retryable,
    }


def _parse_public_adapter_error(
    raw: bytes, *, profile: str, returncode: int
) -> dict[str, Any]:
    fallback = _generic_adapter_error(
        profile,
        category="provider_response",
        code=f"adapter_exit_{returncode}",
        retryable=False,
    )
    if len(raw) > MAX_ADAPTER_RESPONSE_BYTES:
        return fallback
    try:
        document = json.loads(raw.decode("utf-8"))
    except (UnicodeError, json.JSONDecodeError):
        return fallback
    error = document.get("error") if isinstance(document, Mapping) else None
    if not isinstance(error, Mapping) or set(error) - {
        "provider",
        "category",
        "code",
        "http_status",
        "request_id",
        "retryable",
    }:
        return fallback
    provider = error.get("provider")
    category = error.get("category")
    code = error.get("code")
    retryable = error.get("retryable")
    if (
        provider != profile
        or category not in PUBLIC_ERROR_CATEGORIES
        or not isinstance(code, str)
        or PUBLIC_ERROR_TOKEN_RE.fullmatch(code) is None
        or not isinstance(retryable, bool)
    ):
        return fallback
    result: dict[str, Any] = {
        "provider": provider,
        "category": category,
        "code": code,
        "retryable": retryable,
    }
    status = error.get("http_status")
    if status is not None:
        if not isinstance(status, int) or isinstance(status, bool) or not 100 <= status <= 599:
            return fallback
        result["http_status"] = status
    request_id = error.get("request_id")
    if request_id is not None:
        if (
            not isinstance(request_id, str)
            or PUBLIC_ERROR_TOKEN_RE.fullmatch(request_id) is None
        ):
            return fallback
        result["request_id"] = request_id
    return result


def _run_adapter(command: list[str], timeout: int, payload: Mapping[str, Any], root: Path) -> dict[str, Any]:
    with tempfile.TemporaryFile() as stdout, tempfile.TemporaryFile() as stderr:
        try:
            completed = subprocess.run(
                command,
                input=_canonical(payload),
                stdout=stdout,
                stderr=stderr,
                cwd=root,
                timeout=timeout,
                check=False,
            )
        except subprocess.TimeoutExpired as exc:
            raise AdapterError(
                "adapter timed out; confirmation was consumed",
                public_error=_generic_adapter_error(
                    str(payload["adapter"]),
                    category="timeout",
                    code="adapter_timeout",
                    retryable=True,
                ),
            ) from exc
        except OSError as exc:
            raise AdapterError(
                "adapter could not be started; confirmation was consumed",
                public_error=_generic_adapter_error(
                    str(payload["adapter"]),
                    category="configuration",
                    code="adapter_start_failed",
                    retryable=False,
                ),
            ) from exc
        if completed.returncode != 0:
            size = stdout.tell()
            stdout.seek(0)
            raw_error = stdout.read(MAX_ADAPTER_RESPONSE_BYTES + 1)
            public_error = _parse_public_adapter_error(
                raw_error if size <= MAX_ADAPTER_RESPONSE_BYTES else b"",
                profile=str(payload["adapter"]),
                returncode=completed.returncode,
            )
            raise AdapterError(
                f"adapter exited with code {completed.returncode}; confirmation was consumed",
                public_error=public_error,
            )
        size = stdout.tell()
        if size > MAX_ADAPTER_RESPONSE_BYTES:
            raise AdapterError("adapter response is too large; confirmation was consumed")
        stdout.seek(0)
        try:
            response = json.loads(stdout.read().decode("utf-8"))
        except (UnicodeError, json.JSONDecodeError) as exc:
            raise AdapterError("adapter returned invalid JSON; confirmation was consumed") from exc
    if not isinstance(response, dict):
        raise AdapterError("adapter response must be an object; confirmation was consumed")
    return response


def _validate_adapter_outputs(
    job: Mapping[str, Any], response: Mapping[str, Any], output_root: Path
) -> list[tuple[str, Path]]:
    entries = response.get("outputs")
    if not isinstance(entries, list) or not all(isinstance(entry, Mapping) for entry in entries):
        raise AdapterError("adapter outputs are invalid; confirmation was consumed")
    result: list[tuple[str, Path]] = []
    for entry in entries:
        if set(entry) != {"target", "source"}:
            raise AdapterError("adapter output fields are invalid; confirmation was consumed")
        target = entry.get("target")
        source = entry.get("source")
        if not isinstance(target, str) or not isinstance(source, str):
            raise AdapterError("adapter output paths are invalid; confirmation was consumed")
        path = Path(source)
        if not path.is_absolute() or path.parent != output_root or path.name in {"", ".", ".."}:
            raise AdapterError(
                "adapter output must use the run staging directory; confirmation was consumed"
            )
        result.append((target, path))
    expected = list(job["outputs"])
    if [target for target, _ in result] != expected:
        raise AdapterError("adapter outputs do not match the confirmed targets; confirmation was consumed")
    return result


def _copy_output(
    source: Path, target: Path, *, overwrite: bool
) -> tuple[str, int]:
    temporary = target.with_name(f".{target.name}.{uuid.uuid4().hex}.tmp")
    target.parent.mkdir(parents=True, exist_ok=True)
    digest = hashlib.sha256()
    size = 0
    descriptor = -1
    try:
        try:
            before = source.lstat()
        except OSError as exc:
            raise AdapterError(
                "adapter output file is missing; confirmation was consumed"
            ) from exc
        if _is_link_or_reparse(before) or not stat.S_ISREG(before.st_mode):
            raise AdapterError("adapter output file is unsafe; confirmation was consumed")
        descriptor = os.open(source, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
        opened = os.fstat(descriptor)
        if (
            not stat.S_ISREG(opened.st_mode)
            or (opened.st_dev, opened.st_ino) != (before.st_dev, before.st_ino)
        ):
            raise AdapterError("adapter output changed while opening; confirmation was consumed")
        if opened.st_size > MAX_OUTPUT_BYTES:
            raise AdapterError("adapter output file is too large; confirmation was consumed")
        with os.fdopen(descriptor, "rb", closefd=True) as incoming, temporary.open(
            "xb"
        ) as outgoing:
            descriptor = -1
            for chunk in iter(lambda: incoming.read(1024 * 1024), b""):
                size += len(chunk)
                if size > MAX_OUTPUT_BYTES:
                    raise AdapterError("adapter output exceeded the size limit")
                digest.update(chunk)
                outgoing.write(chunk)
            outgoing.flush()
            os.fsync(outgoing.fileno())
        if overwrite:
            os.replace(temporary, target)
        else:
            try:
                os.link(temporary, target, follow_symlinks=False)
            except FileExistsError as exc:
                raise FileExistsError(
                    f"output appeared while production was running: {target.name}"
                ) from exc
            temporary.unlink()
    finally:
        if descriptor >= 0:
            os.close(descriptor)
        try:
            temporary.unlink()
        except FileNotFoundError:
            pass
    return digest.hexdigest(), size


def _latest_run(root: Path, job_id: str) -> dict[str, Any] | None:
    history = _read_run_history(root, job_id)
    return history[-1] if history else None


def _parse_run_timestamp(value: object, *, label: str) -> datetime:
    if not isinstance(value, str) or not value:
        raise ValueError(f"production run {label} is invalid")
    normalized = value[:-1] + "+00:00" if value.endswith("Z") else value
    try:
        parsed = datetime.fromisoformat(normalized)
    except ValueError as exc:
        raise ValueError(f"production run {label} is invalid") from exc
    if parsed.tzinfo is None or parsed.utcoffset() is None:
        raise ValueError(f"production run {label} is invalid")
    return parsed.astimezone(timezone.utc)


def _completed_run_order(run: Mapping[str, Any]) -> tuple[datetime, str]:
    return (
        _parse_run_timestamp(run.get("finished_at"), label="finished_at"),
        str(run["run_id"]),
    )


def _read_run_history(root: Path, job_id: str) -> list[dict[str, Any]]:
    directory_parts = ("runs", _job_key(job_id))
    history: list[dict[str, Any]] = []
    for name in _metadata_json_names(root, directory_parts):
        document = _metadata_read_json(
            root, directory_parts, name, maximum=MAX_RUN_RECORD_BYTES
        )
        if (
            not isinstance(document, dict)
            or document.get("job_id") != job_id
            or not isinstance(document.get("run_id"), str)
            or document.get("status") not in {"running", "succeeded", "failed"}
            or not isinstance(document.get("fingerprint"), str)
            or re.fullmatch(r"[0-9a-f]{64}", str(document.get("fingerprint"))) is None
        ):
            raise ValueError("production run record is invalid")
        started_at = _parse_run_timestamp(
            document.get("started_at"), label="started_at"
        )
        finished_at = document.get("finished_at")
        if document["status"] == "running":
            if finished_at is not None:
                raise ValueError("running production run has finished_at")
        else:
            co

#!/usr/bin/env python3
"""Stdlib-only production adapters for supported media providers.

The public ``compile_*`` functions are deterministic and perform no I/O.  The
CLI reads the confirmed production job from stdin and writes only the adapter
contract JSON to stdout after a provider result has been saved to a temporary
regular file.
"""

from __future__ import annotations

import argparse
import base64
import binascii
import json
import mimetypes
import os
import re
import secrets
import stat
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from collections.abc import Collection, Mapping, Sequence
from pathlib import Path
from typing import Any

OPENAI_MODEL = "gpt-image-2"
MINIMAX_MUSIC_MODEL = "music-3.0"
OPENAI_BASE_URL = "https://api.openai.com/v1"
SEEDANCE_BASE_URL = "https://ark.cn-beijing.volces.com/api/v3"
MINIMAX_BASE_URL = "https://api.minimax.io/v1"
MINIMAX_VIDEO_BASE_URL = "https://api.minimax.io/v2"
MAX_JSON_RESPONSE = 128 * 1024 * 1024
MAX_OUTPUT_BYTES = 512 * 1024 * 1024
MAX_REFERENCE_BYTES = 50 * 1024 * 1024
MAX_MULTIPART_BYTES = 200 * 1024 * 1024
MAX_ERROR_BODY_BYTES = 64 * 1024
TERMINAL_FAILURES = {"failed", "cancelled", "canceled", "timeout", "expired"}
SEEDANCE_RATIOS = {"adaptive", "1:1", "3:4", "4:3", "9:16", "16:9", "21:9"}
SEEDANCE_REFERENCE_ROLES = {
    "first_frame": "image_url",
    "last_frame": "image_url",
    "reference_image": "image_url",
    "reference_video": "video_url",
    "reference_audio": "audio_url",
}
# One opening frame and one closing frame, because each names a single position
# in the take. Note what is deliberately absent: the sibling adapter forbids
# mixing frame conditioning with reference conditioning, and that rule belongs
# to that provider's reference, not to this one. Do not carry it across.
SEEDANCE_SINGULAR_ROLES = {"first_frame", "last_frame"}
MINIMAX_VIDEO_RATIOS = {"adaptive", "1:1", "3:4", "4:3", "9:16", "16:9", "21:9"}
MINIMAX_VIDEO_RESOLUTIONS = {"480P", "768P", "2K"}
MINIMAX_VIDEO_PROMPT_LIMIT = 7000
MINIMAX_SPEECH_TEXT_LIMIT = 5000
MINIMAX_VIDEO_ROLES = {
    "first_frame": "image_url",
    "last_frame": "image_url",
    "reference_image": "image_url",
    "reference_video": "video_url",
    "reference_audio": "audio_url",
}
# Both providers document a `data:<mime>;base64,<...>` URI as an accepted media
# input alongside a public URL, so a local project reference needs no upload
# service to reach them. The caps are MiniMax's published per-modality limits;
# Seedance publishes no numbers, so the same conservative guard is applied there
# and can be raised per deployment. Base64 inflates bytes by about a third, so
# the request cap is checked against the encoded size.
INLINE_REFERENCE_LIMITS = {
    "image_url": 30 * 1024 * 1024,
    "video_url": 50 * 1024 * 1024,
    "audio_url": 15 * 1024 * 1024,
}
INLINE_REFERENCE_BODY_LIMIT = 64 * 1024 * 1024
INLINE_REFERENCE_MIME = {
    ".png": "image/png",
    ".jpg": "image/jpeg",
    ".jpeg": "image/jpeg",
    ".webp": "image/webp",
    ".mp4": "video/mp4",
    ".mov": "video/quicktime",
    ".webm": "video/webm",
    ".wav": "audio/wav",
    ".mp3": "audio/mpeg",
    ".m4a": "audio/mp4",
    ".aac": "audio/aac",
    ".flac": "audio/flac",
}
GPT_IMAGE_MIN_PIXELS = 655_360
GPT_IMAGE_MAX_PIXELS = 8_294_400
MINIMUM_PYTHON = (3, 9)
if sys.version_info < MINIMUM_PYTHON:
    raise SystemExit("provider_adapters.py requires Python 3.9 or newer")


class AdapterFailure(RuntimeError):
    """A safe-to-report adapter failure without provider response contents."""

    def __init__(
        self,
        message: str,
        *,
        category: str = "provider_response",
        code: str = "adapter_failure",
        http_status: int | None = None,
        request_id: str | None = None,
        retryable: bool = False,
    ) -> None:
        super().__init__(message)
        self.category = category
        self.code = _safe_token(code) or "adapter_failure"
        self.http_status = http_status
        self.request_id = _safe_token(request_id)
        self.retryable = retryable

    def public(self, provider: str) -> dict[str, Any]:
        result: dict[str, Any] = {
            "provider": provider,
            "category": self.category,
            "code": self.code,
            "retryable": self.retryable,
        }
        if self.http_status is not None:
            result["http_status"] = self.http_status
        if self.request_id is not None:
            result["request_id"] = self.request_id
        return result


def _safe_token(value: object) -> str | None:
    if isinstance(value, str) and re.fullmatch(
        r"[A-Za-z0-9][A-Za-z0-9._:-]{0,199}", value
    ):
        return value
    if isinstance(value, int) and not isinstance(value, bool):
        return str(value)
    return None


def _request_id(headers: Mapping[str, Any]) -> str | None:
    folded = {str(key).casefold(): value for key, value in headers.items()}
    for name in ("x-request-id", "x-tt-logid", "x-trace-id", "trace-id"):
        value = _safe_token(folded.get(name))
        if value is not None:
            return value
    return None


def _provider_code(document: object) -> str | None:
    if not isinstance(document, Mapping):
        return None
    error = document.get("error")
    if isinstance(error, Mapping):
        value = _safe_token(error.get("code"))
        if value is not None:
            return value
    base_resp = document.get("base_resp")
    if isinstance(base_resp, Mapping):
        value = _safe_token(base_resp.get("status_code"))
        if value not in {None, "0"}:
            return value
    return _safe_token(document.get("code"))


def _http_failure(provider: str, error: urllib.error.HTTPError) -> AdapterFailure:
    status = error.code
    try:
        raw = error.read(MAX_ERROR_BODY_BYTES + 1)
        document = json.loads(raw) if len(raw) <= MAX_ERROR_BODY_BYTES else None
    except (OSError, UnicodeError, json.JSONDecodeError):
        document = None
    if status == 401:
        category = "authentication"
    elif status == 403:
        category = "permission"
    elif status == 429:
        category = "rate_limit"
    elif 500 <= status <= 599:
        category = "server"
    elif 400 <= status <= 499:
        category = "invalid_request"
    else:
        category = "provider_response"
    return AdapterFailure(
        f"{provider} HTTP request failed",
        category=category,
        code=_provider_code(document) or f"http_{status}",
        http_status=status,
        request_id=_request_id(
            dict(error.headers.items()) if error.headers is not None else {}
        ),
        retryable=status == 429 or 500 <= status <= 599,
    )


def _require_job(job: Mapping[str, Any], modality: str) -> tuple[str, dict[str, Any]]:
    if not isinstance(job, Mapping) or job.get("modality") != modality:
        raise ValueError(f"adapter requires a {modality} job")
    prompt = job.get("prompt")
    if not isinstance(prompt, str) or not prompt.strip():
        raise ValueError("job prompt must be non-empty")
    parameters = job.get("parameters", {})
    if not isinstance(parameters, Mapping):
        raise ValueError("job parameters must be an object")
    outputs = job.get("outputs")
    if not isinstance(outputs, list) or len(outputs) != 1 or not isinstance(outputs[0], str):
        raise ValueError("provider adapter requires exactly one output")
    return prompt, dict(parameters)


def _prompt_with_reference_contract(
    prompt: str,
    job: Mapping[str, Any],
    *,
    prompt_language: str | None = None,
    reference_tokens: Sequence[str] = (),
    zh_reference_prefix: str = "参考",
    zh_heading: str = "参考约束：",
) -> str:
    bindings = job.get("reference_bindings", [])
    references = job.get("references", [])
    if not bindings:
        return prompt
    if (
        not isinstance(bindings, list)
        or not isinstance(references, list)
        or len(bindings) != len(references)
    ):
        raise ValueError("reference bindings must match job references")
    if reference_tokens and len(reference_tokens) != len(references):
        raise ValueError("reference tokens must match job references")
    language = (prompt_language or "en").casefold()
    instructions: list[str] = []
    for index, binding in enumerate(bindings, 1):
        if not isinstance(binding, Mapping) or binding.get("order") != index:
            raise ValueError("reference binding order is invalid")
        if binding.get("path") != references[index - 1]:
            raise ValueError("reference binding path does not match job references")
        label = binding.get("label")
        role = binding.get("role")
        may_control = binding.get("may_control")
        must_not_control = binding.get("must_not_control")
        if (
            not isinstance(label, str)
            or not label.strip()
            or not isinstance(role, str)
            or not role.strip()
            or not isinstance(may_control, list)
            or not may_control
            or not all(isinstance(item, str) and item.strip() for item in may_control)
            or not isinstance(must_not_control, list)
            or not must_not_control
            or not all(isinstance(item, str) and item.strip() for item in must_not_control)
        ):
            raise ValueError("reference binding semantics are invalid")
        values = {
            "order": index,
            "reference": reference_tokens[index - 1] if reference_tokens else str(index),
            "label": label.strip(),
            "role": role.strip(),
            "may": ", ".join(item.strip() for item in may_control),
            "must": ", ".join(item.strip() for item in must_not_control),
        }
        if language.startswith("zh"):
            instructions.append(
                f"{zh_reference_prefix} "
                + (
                    "{reference}（{label}），用途 {role}。允许控制：{may}。"
                    "不得控制：{must}。".format(**values)
                )
            )
        elif language.startswith("en"):
            instructions.append(
                "Reference {reference} ({label}), role {role}. May control: {may}. "
                "Must not control: {must}.".format(**values)
            )
        else:
            instructions.append(
                "[REF {order} | {label} | {role}] [+] {may} [-] {must}".format(
                    **values
                )
            )
    if language.startswith("zh"):
        heading = zh_heading
    elif language.startswith("en"):
        heading = "Reference contract:"
    else:
        heading = "<REF_CONTRACT>"
    return f"{prompt}\n\n{heading}\n" + "\n".join(instructions)


def _pop_prompt_language(parameters: dict[str, Any]) -> str | None:
    value = parameters.pop("prompt_language", None)
    if value is None:
        return None
    if not isinstance(value, str) or not value.strip() or len(value) > 64:
        raise ValueError("prompt_language must be a non-empty bounded language tag")
    return value.strip()


def _take(parameters: dict[str, Any], allowed: set[str]) -> dict[str, Any]:
    unknown = set(parameters) - allowed
    if unknown:
        raise ValueError("unsupported provider parameters: " + ", ".join(sorted(unknown)))
    return parameters


def compile_seedance_payload(
    job: Mapping[str, Any],
    *,
    model: str,
    reference_urls: Sequence[str] = (),
    reference_roles: Sequence[str] = (),
    allowed_ratios: Collection[str] | None = None,
    duration_range: tuple[int, int] | None = None,
) -> dict[str, Any]:
    """Compile a video production job into the official Seedance task body.

    ``model`` is deliberately mandatory: callers must obtain it from explicit
    runtime configuration rather than assuming any Seedance release.
    """
    prompt, parameters = _require_job(job, "video")
    if not isinstance(model, str) or not model.strip():
        raise ValueError("Seedance model must be explicitly configured")
    references = job.get("references", [])
    if not isinstance(references, list) or len(reference_urls) != len(references):
        raise ValueError("Seedance reference URLs must match job references")
    if len(reference_roles) != len(reference_urls):
        raise ValueError("Seedance reference roles must match job references")
    if Path(job["outputs"][0]).suffix.casefold() != ".mp4":
        raise ValueError("Seedance adapter requires an MP4 target")
    parameters = _take(
        parameters,
        {
            "duration",
            "ratio",
            "generate_audio",
            "omni_reference_task_type",
            "prompt_language",
        },
    )
    prompt_language = _pop_prompt_language(parameters)
    duration = parameters.get("duration")
    if duration is not None and (
        not isinstance(duration, int)
        or isinstance(duration, bool)
        or (duration != -1 and not 1 <= duration <= 30)
    ):
        raise ValueError("Seedance duration must be -1 or an integer from 1 to 30")
    if duration is not None and duration != -1:
        if duration_range is None:
            raise ValueError("Seedance duration needs an explicit model profile")
        minimum, maximum = duration_range
        if not 1 <= minimum <= maximum <= 30 or not minimum <= duration <= maximum:
            raise ValueError("Seedance duration is outside the configured model profile")
    ratio = parameters.get("ratio")
    if ratio is not None and (
        not isinstance(ratio, str) or ratio.strip() not in SEEDANCE_RATIOS
    ):
        raise ValueError("Seedance ratio is outside the supported profile")
    if ratio is not None:
        configured_ratios = set(allowed_ratios or ())
        if not configured_ratios:
            raise ValueError("Seedance ratio needs an explicit model profile")
        if not configured_ratios <= SEEDANCE_RATIOS or ratio.strip() not in configured_ratios:
            raise ValueError("Seedance ratio is outside the configured model profile")
        ratio = ratio.strip()
    generate_audio = parameters.get("generate_audio")
    if generate_audio is not None and not isinstance(generate_audio, bool):
        raise ValueError("Seedance generate_audio must be a boolean")
    task_type = parameters.get("omni_reference_task_type")
    if task_type is not None and task_type not in {"auto", "reference", "edit", "extend"}:
        raise ValueError("Seedance omni reference task type is invalid")
    counters = {"image_url": 0, "video_url": 0, "audio_url": 0}
    labels = {"image_url": "图片", "video_url": "视频", "audio_url": "音频"}
    reference_tokens: list[str] = []
    singular_seen: set[str] = set()
    for role in reference_roles:
        field = SEEDANCE_REFERENCE_ROLES.get(role)
        if field is None:
            raise ValueError(f"unsupported Seedance reference role: {role}")
        if role in SEEDANCE_SINGULAR_ROLES:
            if role in singular_seen:
                raise ValueError(f"Seedance accepts one {role} reference")
            singular_seen.add(role)
        # A frame role still occupies a picture slot, so it keeps counting: the
        # tokens are positional, and skipping one would point every later
        # `@图片N` in the prose at the wrong picture.
        counters[field] += 1
        reference_tokens.append(f"@{labels[field]}{counters[field]}")
    if task_type in {"edit", "extend"} and "reference_video" not in reference_roles:
        raise ValueError(f"Seedance {task_type} requires a reference video")
    if task_type == "edit" and (ratio != "adaptive" or duration != -1):
        raise ValueError("Seedance edit requires adaptive ratio and duration -1")
    if task_type == "extend" and ratio != "adaptive":
        raise ValueError("Seedance extend requires an adaptive ratio")
    text = _prompt_with_reference_contract(
        prompt,
        job,
        prompt_language=prompt_language,
        reference_tokens=reference_tokens,
        zh_reference_prefix=(
            "输入素材" if task_type in {"edit", "extend"} else "参考"
        ),
        zh_heading=(
            "输入素材约束：" if task_type in {"edit", "extend"} else "参考约束："
        ),
    )
    content: list[dict[str, Any]] = [{"type": "text", "text": text}]
    for index, url in enumerate(reference_urls):
        if not isinstance(url, str) or not url:
            raise ValueError("Seedance reference URL must be non-empty")
        role = reference_roles[index]
        field = SEEDANCE_REFERENCE_ROLES[role]
        suffix = Path(str(references[index])).suffix.casefold()
        expected = {
            "image_url": {".png", ".jpg", ".jpeg", ".webp"},
            "video_url": {".mp4", ".mov", ".webm"},
            "audio_url": {".wav", ".mp3", ".m4a", ".aac", ".flac"},
        }[field]
        if suffix not in expected:
            raise ValueError(f"Seedance {role} does not match the reference file type")
        parsed = urllib.parse.urlparse(url)
        if not (
            (parsed.scheme == "https" and parsed.netloc)
            or (
                parsed.scheme == "asset"
                and parsed.netloc.startswith("asset-")
                and not parsed.path
            )
            or _is_inline_reference(url)
        ):
            raise ValueError(
                "Seedance reference URL must be HTTPS, asset:// or a base64 data URI"
            )
        content.append({"type": field, field: {"url": url}, "role": role})
    body: dict[str, Any] = {"model": model.strip(), "content": content}
    if ratio is not None:
        body["ratio"] = ratio
    if duration is not None:
        body["duration"] = duration
    if generate_audio is not None:
        body["generate_audio"] = generate_audio
    if task_type is not None:
        body["omni_reference_task_type"] = task_type
    return body


def _seedance_runtime_profile(
) -> tuple[frozenset[str] | None, tuple[int, int] | None]:
    raw_ratios = os.environ.get("SEEDANCE_ALLOWED_RATIOS")
    allowed_ratios = (
        frozenset(item.strip() for item in raw_ratios.split(",") if item.strip())
        if raw_ratios is not None
        else None
    )
    minimum_raw = os.environ.get("SEEDANCE_MIN_DURATION")
    maximum_raw = os.environ.get("SEEDANCE_MAX_DURATION")
    if (minimum_raw is None) != (maximum_raw is None):
        raise AdapterFailure(
            "Seedance duration profile is incomplete",
            category="configuration",
            code="invalid_model_profile",
        )
    try:
        duration_range = (
            (int(minimum_raw), int(maximum_raw))
            if minimum_raw is not None and maximum_raw is not None
            else None
        )
    except ValueError as exc:
        raise AdapterFailure(
            "Seedance duration profile is invalid",
            category="configuration",
            code="invalid_model_profile",
        ) from exc
    if allowed_ratios is not None and (
        not allowed_ratios or not allowed_ratios <= SEEDANCE_RATIOS
    ):
        raise AdapterFailure(
            "Seedance ratio profile is invalid",
            category="configuration",
            code="invalid_model_profile",
        )
    if duration_range is not None and not (
        1 <= duration_range[0] <= duration_range[1] <= 30
    ):
        raise AdapterFailure(
            "Seedance duration profile is invalid",
            category="configuration",
            code="invalid_model_profile",
        )
    return allowed_ratios, duration_range


def compile_minimax_h3_payload(
    job: Mapping[str, Any],
    *,
    model: str,
    reference_urls: Sequence[str] = (),
    reference_roles: Sequence[str] = (),
    allowed_ratios: Collection[str] | None = None,
    allowed_resolutions: Collection[str] | None = None,
    duration_range: tuple[int, int] | None = None,
) -> dict[str, Any]:
    """Compile a video job into the official MiniMax video-generation task body.

    ``model`` is mandatory for the same reason it is for Seedance: which release
    is enabled, and what it accepts, is deployment configuration, never an
    assumption made here.
    """
    prompt, parameters = _require_job(job, "video")
    if not isinstance(model, str) or not model.strip():
        raise ValueError("MiniMax video model must be explicitly configured")
    references = job.get("references", [])
    if not isinstance(references, list) or len(reference_urls) != len(references):
        raise ValueError("MiniMax reference URLs must match job references")
    if len(reference_roles) != len(reference_urls):
        raise ValueError("MiniMax reference roles must match job references")
    if Path(job["outputs"][0]).suffix.casefold() != ".mp4":
        raise ValueError("MiniMax video adapter requires an MP4 target")
    parameters = _take(
        parameters, {"duration", "ratio", "resolution", "prompt_language"}
    )
    prompt_language = _pop_prompt_language(parameters)

    duration = parameters.get("duration")
    if not isinstance(duration, int) or isinstance(duration, bool):
        raise ValueError("MiniMax video duration must be an integer number of seconds")
    if duration_range is None:
        raise ValueError("MiniMax video duration needs an explicit model profile")
    minimum, maximum = duration_range
    if not 1 <= minimum <= maximum or not minimum <= duration <= maximum:
        raise ValueError("MiniMax video duration is outside the configured model profile")

    resolution = parameters.get("resolution")
    if not isinstance(resolution, str) or resolution.strip() not in MINIMAX_VIDEO_RESOLUTIONS:
        raise ValueError("MiniMax video resolution is outside the supported profile")
    configured_resolutions = set(allowed_resolutions or ())
    if not configured_resolutions:
        raise ValueError("MiniMax video resolution needs an explicit model profile")
    if (
        not configured_resolutions <= MINIMAX_VIDEO_RESOLUTIONS
        or resolution.strip() not in configured_resolutions
    ):
        raise ValueError("MiniMax video resolution is outside the configured model profile")

    ratio = parameters.get("ratio")
    if ratio is not None:
        if not isinstance(ratio, str) or ratio.strip() not in MINIMAX_VIDEO_RATIOS:
            raise ValueError("MiniMax video ratio is outside the supported profile")
        configured_ratios = set(allowed_ratios or ())
        if not configured_ratios:
            raise ValueError("MiniMax video ratio needs an explicit model profile")
        if (
            not configured_ratios <= MINIMAX_VIDEO_RATIOS
            or ratio.strip() not in configured_ratios
        ):
            raise ValueError("MiniMax video ratio is outside the configured model profile")

    counters = {"image_url": 0, "video_url": 0, "audio_url": 0}
    labels = {"image_url": "Picture", "video_url": "Video", "audio_url": "Audio"}
    reference_tokens: list[str] = []
    for role in reference_roles:
        field = MINIMAX_VIDEO_ROLES.get(role)
        if field is None:
            raise ValueError(f"unsupported MiniMax reference role: {role}")
        counters[field] += 1
        reference_tokens.append(f"<{labels[field]} {counters[field]}>")
    text = _prompt_with_reference_contract(
        prompt,
        job,
        prompt_language=prompt_language,
        reference_tokens=reference_tokens,
    )
    if len(text) > MINIMAX_VIDEO_PROMPT_LIMIT:
        raise ValueError("MiniMax video prompt exceeds the provider limit")
    content: list[dict[str, Any]] = [{"type": "text", "text": text}]
    seen_roles: list[str] = []
    for index, url in enumerate(reference_urls):
        role = reference_roles[index]
        if role not in MINIMAX_VIDEO_ROLES:
            raise ValueError(f"unsupported MiniMax reference role: {role}")
        if role in {"first_frame", "last_frame"} and role in seen_roles:
            raise ValueError(f"MiniMax accepts one {role} reference")
        seen_roles.append(role)
        if not isinstance(url, str) or not url:
            raise ValueError("MiniMax reference URL must be non-empty")
        parsed = urllib.parse.urlparse(url)
        if not (
            (parsed.scheme == "https" and parsed.netloc)
            or (parsed.scheme == "mm_file" and parsed.netloc and not parsed.path)
            or _is_inline_reference(url)
        ):
            raise ValueError(
                "MiniMax reference URL must be HTTPS, mm_file:// or a base64 data URI"
            )
        field = MINIMAX_VIDEO_ROLES[role]
        content.append({"type": field, field: {"url": url}, "role": role})
    frame_roles = {"first_frame", "last_frame"}
    reference_roles_present = {"reference_image", "reference_video", "reference_audio"}
    if frame_roles.intersection(seen_roles) and reference_roles_present.intersection(
        seen_roles
    ):
        raise ValueError(
            "MiniMax frame conditioning cannot be mixed with reference conditioning"
        )
    if ratio is None and len(content) == 1:
        raise ValueError("MiniMax text-to-video requires an explicit ratio")
    body: dict[str, Any] = {
        "model": model.strip(),
        "content": content,
        "duration": duration,
        "resolution": resolution.strip(),
    }
    if ratio is not None:
        if ratio.strip() == "adaptive" and len(content) == 1:
            raise ValueError("MiniMax text-to-video cannot use an adaptive ratio")
        body["ratio"] = ratio.strip()
    return body


def _minimax_video_runtime_profile() -> tuple[
    frozenset[str] | None, frozenset[str] | None, tuple[int, int] | None
]:
    def _set(name: str, supported: Collection[str]) -> frozenset[str] | None:
        raw = os.environ.get(name)
        if raw is None:
            return None
        values = frozenset(item.strip() for item in raw.split(",") if item.strip())
        if not values or not values <= set(supported):
            raise AdapterFailure(
                f"MiniMax video profile is invalid: {name}",
                category="configuration",
                code="invalid_model_profile",
            )
        return values

    allowed_ratios = _set("MINIMAX_VIDEO_RATIOS", MINIMAX_VIDEO_RATIOS)
    allowed_resolutions = _set("MINIMAX_VIDEO_RESOLUTIONS", MINIMAX_VIDEO_RESOLUTIONS)
    minimum_raw = os.environ.get("MINIMAX_VIDEO_MIN_DURATION")
    maximum_raw = os.environ.get("MINIMAX_VIDEO_MAX_DURATION")
    if (minimum_raw is None) != (maximum_raw is None):
        raise AdapterFailure(
            "MiniMax video duration profile is incomplete",
            category="configuration",
            code="invalid_model_profile",
        )
    try:
        duration_range = (
            (int(minimum_raw), int(maximum_raw))
            if minimum_raw is not None and maximum_raw is not None
            else None
        )
    except ValueError as exc:
        raise AdapterFailure(
            "MiniMax video duration profile is invalid",
            category="configuration",
            code="invalid_model_profile",
        ) from exc
    if duration_range is not None and not 1 <= duration_range[0] <= duration_range[1]:
        raise AdapterFailure(
            "MiniMax video duration profile is invalid",
            category="configuration",
            code="invalid_model_profile",
        )
    return allowed_ratios, allowed_resolutions, duration_range


def compile_gpt_image_2_payload(job: Mapping[str, Any]) -> dict[str, Any]:
    """Compile an image job into GPT Image 2 generation/edit fields."""
    prompt, parameters = _require_job(job, "image")
    parameters = _take(
        parameters,
        {
            "width",
            "height",
            "size",
            "quality",
            "background",
            "moderation",
            "prompt_language",
        },
    )
    prompt_language = _pop_prompt_language(parameters)
    prompt = _prompt_with_reference_contract(
        prompt, job, prompt_language=prompt_language
    )
    if len(prompt) > 32000:
        raise ValueError("GPT Image prompt exceeds the provider limit")
    width = parameters.pop("width", None)
    height = parameters.pop("height", None)
    if (width is None) != (height is None) or (width is not None and "size" in parameters):
        raise ValueError("use either both width/height or size")
    if width is not None:
        if not isinstance(width, int) or not isinstance(height, int) or width <= 0 or height <= 0:
            raise ValueError("image dimensions must be positive integers")
        if width % 16 or height % 16 or not (1 / 3 <= width / height <= 3):
            raise ValueError("GPT Image dimensions must be divisible by 16 with ratio between 1:3 and 3:1")
        parameters["size"] = f"{width}x{height}"
    size = parameters.get("size")
    if size != "auto" and size is not None:
        if not isinstance(size, str) or "x" not in size:
            raise ValueError("GPT Image size is invalid")
        try:
            parsed_width, parsed_height = (int(part) for part in size.split("x", 1))
        except ValueError as exc:
            raise ValueError("GPT Image size is invalid") from exc
        if (
            parsed_width <= 0
            or parsed_height <= 0
            or parsed_width % 16
            or parsed_height % 16
            or not (1 / 3 <= parsed_width / parsed_height <= 3)
            or parsed_width > 3840
            or parsed_height > 3840
            or not (
                GPT_IMAGE_MIN_PIXELS
                <= parsed_width * parsed_height
                <= GPT_IMAGE_MAX_PIXELS
            )
        ):
            raise ValueError("GPT Image size violates official dimension constraints")
    if parameters.get("background") == "transparent":
        raise ValueError("GPT Image 2 transparent background is unsupported")
    if "background" in parameters and parameters["background"] not in {"auto", "opaque"}:
        raise ValueError("GPT Image background is invalid")
    if "quality" in parameters and parameters["quality"] not in {"auto", "low", "medium", "high"}:
        raise ValueError("GPT Image quality is invalid")
    if "moderation" in parameters and parameters["moderation"] not in {"auto", "low"}:
        raise ValueError("GPT Image moderation is invalid")
    suffix = Path(job["outputs"][0]).suffix.casefold()
    formats = {".png": "png", ".jpg": "jpeg", ".jpeg": "jpeg", ".webp": "webp"}
    if suffix not in formats:
        raise ValueError("unsupported GPT Image output extension")
    references = job.get("references", [])
    if not isinstance(references, list) or len(references) > 16:
        raise ValueError("GPT Image accepts at most sixteen references")
    if any(
        not isinstance(reference, str)
        or Path(reference).suffix.casefold() not in {".png", ".jpg", ".jpeg", ".webp"}
        for reference in references
    ):
        raise ValueError("GPT Image references must be supported image files")
    return {
        "model": OPENAI_MODEL,
        "prompt": prompt,
        "n": 1,
        "output_format": formats[suffix],
        **parameters,
    }


MINIMAX_SPEECH_EMOTIONS = {
    "happy", "sad", "angry", "fearful", "disgusted", "surprised", "neutral",
}


def compile_minimax_speech_payload(job: Mapping[str, Any]) -> dict[str, Any]:
    """Compile a tts job into the official MiniMax speech JSON body.

    The voice catalogue is deliberately not enumerated here. Which preset voices
    an account can reach depends on the model and the account, no published list
    is authoritative for both, and a list frozen into this file would either
    refuse a voice that works or vouch for one that does not. The document owns
    the value and a reviewer reads it; the adapter only checks its shape.
    """

    text, parameters = _require_job(job, "tts")
    parameters = _take(
        parameters,
        {"model", "voice_id", "speed", "vol", "pitch", "emotion",
         "sample_rate", "bitrate", "format"},
    )
    model = parameters.pop("model", "")
    if not isinstance(model, str) or not model.strip():
        raise ValueError("MiniMax speech model must be explicitly configured")
    voice_id = parameters.pop("voice_id", "")
    if not isinstance(voice_id, str) or not voice_id.strip() or voice_id != voice_id.strip():
        raise ValueError("MiniMax speech requires a confirmed voice_id")
    if len(text) > MINIMAX_SPEECH_TEXT_LIMIT:
        raise ValueError("MiniMax speech text exceeds the provider limit")

    target_format = Path(job["outputs"][0]).suffix.casefold().lstrip(".")
    requested_format = parameters.pop("format", target_format)
    if requested_format != target_format or requested_format not in {"mp3", "wav"}:
        raise ValueError("MiniMax output format must match a supported target extension")
    sample_rate = parameters.pop("sample_rate", 32000)
    bitrate = parameters.pop("bitrate", 128000)
    if sample_rate not in {16000, 24000, 32000, 44100}:
        raise ValueError("unsupported MiniMax sample rate")
    if bitrate not in {32000, 64000, 128000, 256000}:
        raise ValueError("unsupported MiniMax bitrate")

    voice_setting: dict[str, Any] = {"voice_id": voice_id}
    emotion = parameters.pop("emotion", None)
    if emotion is not None:
        if emotion not in MINIMAX_SPEECH_EMOTIONS:
            raise ValueError("unsupported MiniMax speech emotion")
        voice_setting["emotion"] = emotion
    for name, low, high in (("speed", 0.5, 2.0), ("vol", 0.1, 10.0)):
        value = parameters.pop(name, None)
        if value is None:
            continue
        if isinstance(value, bool) or not isinstance(value, (int, float)):
            raise ValueError(f"MiniMax speech {name} must be a number")
        if not low <= float(value) <= high:
            raise ValueError(f"MiniMax speech {name} is outside the supported range")
        voice_setting[name] = float(value)
    pitch = parameters.pop("pitch", None)
    if pitch is not None:
        if isinstance(pitch, bool) or not isinstance(pitch, int) or not -12 <= pitch <= 12:
            raise ValueError("MiniMax speech pitch is outside the supported range")
        voice_setting["pitch"] = pitch

    return {
        "model": model.strip(),
        "text": text,
        "stream": False,
        "output_format": "hex",
        "voice_setting": voice_setting,
        "audio_setting": {
            "sample_rate": sample_rate,
            "bitrate": bitrate,
            "format": requested_format,
        },
    }


def compile_minimax_music_payload(job: Mapping[str, Any]) -> dict[str, Any]:
    """Compile an audio job into the official MiniMax Music 3.0 JSON body."""
    prompt, parameters = _require_job(job, "music")
    parameters = _take(
        parameters,
        {"lyrics", "sample_rate", "bitrate", "format", "lyrics_optimizer", "is_instrumental"},
    )
    target_format = Path(job["outputs"][0]).suffix.casefold().lstrip(".")
    requested_format = parameters.pop("format", target_format)
    if requested_format != target_format or requested_format not in {"mp3", "wav"}:
        raise ValueError("MiniMax output format must match a supported target extension")
    lyrics = parameters.pop("lyrics", None)
    instrumental = parameters.pop("is_instrumental", False)
    optimizer = parameters.pop("lyrics_optimizer", False)
    if not isinstance(instrumental, bool) or not isinstance(optimizer, bool):
        raise ValueError("MiniMax boolean parameters must be booleans")
    if optimizer:
        raise ValueError("MiniMax lyrics_optimizer is forbidden for confirmed production")
    if instrumental and lyrics not in {None, ""}:
        raise ValueError("MiniMax instrumental music must not carry lyrics")
    if not instrumental and (not isinstance(lyrics, str) or not lyrics.strip()):
        raise ValueError("MiniMax vocal music requires confirmed lyrics")
    if len(prompt) > 2000 or (isinstance(lyrics, str) and len(lyrics) > 3500):
        raise ValueError("MiniMax prompt or lyrics exceeds the provider limit")
    sample_rate = parameters.pop("sample_rate", 44100)
    bitrate = parameters.pop("bitrate", 256000)
    if sample_rate not in {16000, 24000, 32000, 44100}:
        raise ValueError("unsupported MiniMax sample rate")
    if bitrate not in {32000, 64000, 128000, 256000}:
        raise ValueError("unsupported MiniMax bitrate")
    body: dict[str, Any] = {
        "model": MINIMAX_MUSIC_MODEL,
        "prompt": prompt,
        "stream": False,
        "output_format": "hex",
        "audio_setting": {
            "sample_rate": sample_rate,
            "bitrate": bitrate,
            "format": requested_format,
        },
        "lyrics_optimizer": False,
        "is_instrumental": instrumental,
    }
    if lyrics is not None:
        if not isinstance(lyrics, str):
            raise ValueError("MiniMax lyrics must be a string")
        body["lyrics"] = lyrics
    return body


def _base_url(env_name: str, default: str) -> str:
    value = os.environ.get(env_name, default).rstrip("/")
    parsed = urllib.parse.urlparse(value)
    if parsed.scheme != "https" or not parsed.netloc or parsed.username or parsed.password:
        raise AdapterFailure(
            "provider base URL is invalid",
            category="configuration",
            code="invalid_base_url",
        )
    return value


def _credential(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        raise AdapterFailure(
            "provider credential is not configured",
            category="configuration",
            code="missing_credential",
        )
    return value


def _request_json(
    url: str,
    *,
    provider: str,
    method: str = "POST",
    body: Mapping[str, Any] | None = None,
    token: str,
) -> tuple[dict[str, Any], Mapping[str, str]]:
    data = None if body is None else json.dumps(body, ensure_ascii=False).encode("utf-8")
    request = urllib.request.Request(url, data=data, method=method)
    request.add_header("Authorization", f"Bearer {token}")
    if body is not None:
        request.add_header("Content-Type", "application/json")
    try:
        with urllib.request.urlopen(request, timeout=180) as response:
            raw = response.read(MAX_JSON_RESPONSE + 1)
            headers = dict(response.headers.items())
    except urllib.error.HTTPError as exc:
        raise _http_failure(provider, exc) from exc
    except TimeoutError as exc:
        raise AdapterFailure(
            f"{provider} HTTP request timed out",
            category="timeout",
            code="request_timeout",
            retryable=True,
        ) from exc
    except (urllib.error.URLError, OSError) as exc:
        raise AdapterFailure(
            f"{provider} HTTP request failed",
            category="network",
            code="network_error",
            retryable=True,
        ) from exc
    if len(raw) > MAX_JSON_RESPONSE:
        raise AdapterFailure(
            f"{provider} response is too large", code="response_too_large"
        )
    try:
        document = json.loads(raw)
    except (UnicodeError, json.JSONDecodeError) as exc:
        raise AdapterFailure(
            f"{provider} returned invalid JSON", code="invalid_json"
        ) from exc
    if not isinstance(document, dict):
        raise AdapterFailure(
            f"{provider} returned an invalid response", code="invalid_response"
        )
    return document, headers


def _read_reference(path: Path) -> bytes:
    before = path.lstat()
    if stat.S_ISLNK(before.st_mode) or not stat.S_ISREG(before.st_mode):
        raise ValueError("reference is not a regular file")
    if before.st_size > MAX_REFERENCE_BYTES:
        raise ValueError("reference exceeds the provider input size limit")
    # Windows opens files in text mode unless told otherwise, which stops the
    # read at the first 0x1A byte -- the seventh byte of every PNG signature.
    flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0)
    descriptor = os.open(path, flags)
    try:
        opened = os.fstat(descriptor)
        if (
            not stat.S_ISREG(opened.st_mode)
            or (opened.st_dev, opened.st_ino) != (before.st_dev, before.st_ino)
        ):
            raise ValueError("reference changed while opening")
        chunks: list[bytes] = []
        size = 0
        while chunk := os.read(descriptor, 1024 * 1024):
            size += len(chunk)
            if size > MAX_REFERENCE_BYTES:
                raise ValueError("reference exceeds the provider input size limit")
            chunks.append(chunk)
        return b"".join(chunks)
    finally:
        os.close(descriptor)


def _multipart(fields: Mapping[str, Any], paths: Sequence[Path]) -> tuple[bytes, str]:
    boundary = "short-drama-" + secrets.token_hex(16)
    chunks: list[bytes] = []
    for name, value in fields.items():
        chunks.extend(
            [
                f"--{boundary}\r\n".encode(),
                f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode(),
                str(value).lower().encode() if isinstance(value, bool) else str(value).encode("utf-8"),
                b"\r\n",
            ]
        )
    for path in paths:
        content = _read_reference(path)
        media_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
        chunks.extend(
            [
                f"--{boundary}\r\n".encode(),
                f'Content-Disposition: form-data; name="image[]"; filename="{path.name}"\r\n'.encode(),
                f"Content-Type: {media_type}\r\n\r\n".encode(),
                content,
                b"\r\n",
            ]
        )
    chunks.append(f"--{boundary}--\r\n".encode())
    encoded = b"".join(chunks)
    if len(encoded) > MAX_MULTIPART_BYTES:
        raise ValueError("multipart provider input exceeds the total size limit")
    return encoded, f"multipart/form-data; boundary={boundary}"


def _reference_paths(job: Mapping[str, Any]) -> list[Path]:
    raw_root = job.get("project_root")
    if not isinstance(raw_root, str) or not Path(raw_root).is_absolute():
        raise ValueError("project_root is invalid")
    root = Path(raw_root).resolve()
    if not root.is_dir():
        raise ValueError("project_root is invalid")
    result: list[Path] = []
    for reference in job.get("references", []):
        if not isinstance(reference, str):
            raise ValueError("reference path is invalid")
        path = (root / reference).resolve()
        try:
            path.relative_to(root)
        except ValueError as exc:
            raise ValueError("reference escapes project root") from exc
        if not path.is_file() or path.is_symlink():
            raise ValueError("reference is not a regular file")
        result.append(path)
    return result


def _binding_roles(
    job: Mapping[str, Any], *, allowed: Mapping[str, str], provider: str
) -> list[str]:
    """The provider role of each reference, taken from the confirmed job.

    `role` is the production-side translation of the creator document's 用途, so
    it is the job -- not this adapter -- that decides what a picture is for.
    """
    bindings = job.get("reference_bindings", [])
    references = job.get("references", [])
    if not isinstance(bindings, list) or len(bindings) != len(references):
        raise AdapterFailure(
            f"{provider} references need one reference_bindings entry each, "
            "carrying the provider role for that file",
            category="configuration",
            code="missing_reference_roles",
        )
    roles: list[str] = []
    for binding in bindings:
        role = binding.get("role") if isinstance(binding, Mapping) else None
        if not isinstance(role, str) or role not in allowed:
            raise AdapterFailure(
                f"{provider} reference role must be one of "
                + ", ".join(sorted(allowed))
                + f"; got {role!r}",
                category="configuration",
                code="invalid_reference_role",
            )
        roles.append(role)
    return roles


def _inline_reference_urls(
    paths: Sequence[Path], roles: Sequence[str], *, allowed: Mapping[str, str], provider: str
) -> list[str]:
    """Encode local project references as `data:` URIs the provider accepts."""
    urls: list[str] = []
    total = 0
    for path, role in zip(paths, roles):
        field = allowed[role]
        suffix = path.suffix.casefold()
        mime = INLINE_REFERENCE_MIME.get(suffix)
        expected = field.split("_", 1)[0]
        if mime is None or not mime.startswith(expected):
            raise AdapterFailure(
                f"{provider} {role} does not accept a {suffix or 'suffixless'} file",
                category="configuration",
                code="invalid_reference_type",
            )
        try:
            content = path.read_bytes()
        except OSError as exc:
            raise AdapterFailure(
                f"{provider} could not read a project reference",
                category="configuration",
                code="unreadable_reference",
            ) from exc
        if not content:
            raise AdapterFailure(
                f"{provider} project reference is empty",
                category="configuration",
                code="empty_reference",
            )
        # The bytes have to be the media type the extension claims, or the
        # provider rejects a request this adapter said was well formed.
        _validate_media_content(path.name, content)
        limit = INLINE_REFERENCE_LIMITS[field]
        if len(content) > limit:
            raise AdapterFailure(
                f"{provider} reference exceeds the {limit // (1024 * 1024)}MB inline "
                "limit; host it and bind an HTTPS URL instead",
                category="configuration",
                code="reference_too_large",
            )
        encoded = base64.b64encode(content).decode("ascii")
        total += len(encoded)
        if total > INLINE_REFERENCE_BODY_LIMIT:
            raise AdapterFailure(
                f"{provider} inline references exceed the request body limit; "
                "host the largest ones and bind HTTPS URLs instead",
                category="configuration",
                code="reference_body_too_large",
            )
        urls.append(f"data:{mime};base64,{encoded}")
    return urls


def _is_inline_reference(url: str) -> bool:
    return bool(re.fullmatch(r"data:[\w.+-]+/[\w.+-]+;base64,[A-Za-z0-9+/]+=*", url))


def _validate_media_content(target: str, content: bytes) -> None:
    suffix = Path(target).suffix.casefold()
    signatures = {
        ".png": content.startswith(b"\x89PNG\r\n\x1a\n"),
        ".jpg": content.startswith(b"\xff\xd8\xff"),
        ".jpeg": content.startswith(b"\xff\xd8\xff"),
        ".webp": len(content) >= 12
        and content.startswith(b"RIFF")
        and content[8:12] == b"WEBP",
        ".mp3": content.startswith(b"ID3")
        or (
            len(content) >= 2
            and content[0] == 0xFF
            and content[1] & 0xE0 == 0xE0
        ),
        ".wav": len(content) >= 12
        and content.startswith(b"RIFF")
        and content[8:12] == b"WAVE",
        ".mp4": len(content) >= 12 and content[4:8] == b"ftyp",
    }
    if not signatures.get(suffix, False):
        raise AdapterFailure("provider output does not match the target media type")


def _output_root(job: Mapping[str, Any]) -> Path:
    raw = job.get("output_root")
    if not isinstance(raw, str):
        raise ValueError("output_root is invalid")
    root = Path(raw)
    if not root.is_absolute():
        raise ValueError("output_root is invalid")
    try:
        details = root.lstat()
    except OSError as exc:
        raise ValueError("output_root is missing") from exc
    if stat.S_ISLNK(details.st_mode) or not stat.S_ISDIR(details.st_mode):
        raise ValueError("output_root is unsafe")
    return root


def _temporary_output(
    job: Mapping[str, Any], target: str, content: bytes
) -> Path:
    if not content or len(content) > MAX_OUTPUT_BYTES:
        raise AdapterFailure("provider output is too large")
    _validate_media_content(target, content)
    path = _output_root(job) / ("result" + Path(target).suffix.casefold())
    with path.open("xb") as handle:
        handle.write(content)
    return path


def _download(
    job: Mapping[str, Any],
    url: str,
    target: str,
    *,
    provider: str = "seedance",
) -> Path:
    parsed = urllib.parse.urlparse(url)
    if parsed.scheme != "https" or not parsed.netloc:
        raise AdapterFailure("provider output URL is invalid")
    path = _output_root(job) / ("result" + Path(target).suffix.casefold())
    size = 0
    try:
        with urllib.request.urlopen(url, timeout=180) as response, path.open("xb") as handle:
            while chunk := response.read(1024 * 1024):
                size += len(chunk)
                if size > MAX_OUTPUT_BYTES:
                    raise AdapterFailure("provider output is too large")
                handle.write(chunk)
    except AdapterFailure:
        path.unlink(missing_ok=True)
        raise
    except urllib.error.HTTPError as exc:
        path.unlink(missing_ok=True)
        raise _http_failure(provider, exc) from exc
    except TimeoutError as exc:
        path.unlink(missing_ok=True)
        raise AdapterFailure(
            "provider output download timed out",
            category="timeout",
            code="download_timeout",
            retryable=True,
        ) from exc
    except (urllib.error.URLError, OSError) as exc:
        path.unlink(missing_ok=True)
        raise AdapterFailure(
            "provider output download failed",
            category="network",
            code="download_failed",
            retryable=True,
        ) from exc
    if size == 0:
        raise AdapterFailure("provider output is empty")
    with path.open("rb") as handle:
        _validate_media_content(target, handle.read(16))
    return path


def _record_handle(job: Mapping[str, Any], provider_job_id: str) -> None:
    """Write the provider task id where the caller can find it after a crash.

    A video task is billed at submission. Everything after that — polling,
    downloading — can be interrupted, and without this the caller is left with a
    live, already-paid task it has no id for. Written before the first poll, and
    deliberately best-effort: failing to record the handle must not fail a task
    that was submitted successfully.
    """

    destination = job.get("handle_path")
    if not isinstance(destination, str) or not destination:
        return
    try:
        path = Path(destination)
        path.parent.mkdir(parents=True, exist_ok=True)
        temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
        temporary.write_text(
            json.dumps({"provider_job_id": provider_job_id}, ensure_ascii=True),
            encoding="utf-8",
        )
        temporary.replace(path)
    except OSError:
        return


def _collect_target(job: Mapping[str, Any]) -> str | None:
    value = job.get("collect_provider_job_id")
    if isinstance(value, str) and value:
        return value
    return None


def _run_seedance(job: Mapping[str, Any]) -> tuple[Path, str]:
    token = _credential("ARK_API_KEY")
    base = _base_url("SEEDANCE_BASE_URL", SEEDANCE_BASE_URL)
    collecting = _collect_target(job)
    if collecting is not None:
        return _poll_seedance(job, base=base, token=token, task_id=collecting)
    model = os.environ.get("SEEDANCE_MODEL", "")
    references = _reference_paths(job)
    reference_roles = (
        _binding_roles(job, allowed=SEEDANCE_REFERENCE_ROLES, provider="Seedance")
        if references
        else []
    )
    reference_urls = _inline_reference_urls(
        references, reference_roles, allowed=SEEDANCE_REFERENCE_ROLES, provider="Seedance"
    )
    allowed_ratios, duration_range = _seedance_runtime_profile()
    body = compile_seedance_payload(
        job,
        model=model,
        reference_urls=reference_urls,
        reference_roles=reference_roles,
        allowed_ratios=allowed_ratios,
        duration_range=duration_range,
    )
    created, _ = _request_json(
        f"{base}/contents/generations/tasks",
        provider="seedance",
        body=body,
        token=token,
    )
    task_id = created.get("id")
    if not isinstance(task_id, str) or not task_id:
        raise AdapterFailure(
            "Seedance did not return a task id", code="missing_task_id"
        )
    # Billed from here on. Record the id before the first poll.
    _record_handle(job, task_id)
    return _poll_seedance(job, base=base, token=token, task_id=task_id)


def _poll_seedance(
    job: Mapping[str, Any], *, base: str, token: str, task_id: str
) -> tuple[Path, str]:
    try:
        interval = float(os.environ.get("SEEDANCE_POLL_INTERVAL", "5"))
        deadline = time.monotonic() + float(os.environ.get("SEEDANCE_TIMEOUT_SECONDS", "1800"))
    except ValueError as exc:
        raise AdapterFailure("Seedance polling configuration is invalid") from exc
    if interval <= 0 or deadline <= time.monotonic():
        raise AdapterFailure("Seedance polling configuration is invalid")
    while time.monotonic() < deadline:
        status_doc, _ = _request_json(
            f"{base}/contents/generations/tasks/{urllib.parse.quote(task_id, safe='')}",
            provider="seedance",
            method="GET",
            token=token,
        )
        status = status_doc.get("status")
        if status == "succeeded":
            content = status_doc.get("content")
            url = content.get("video_url") if isinstance(content, Mapping) else None
            if not isinstance(url, str):
                raise AdapterFailure(
                    "Seedance succeeded without a video URL",
                    code="missing_video_url",
                    request_id=task_id,
                )
            return _download(
                job,
                url,
                job["outputs"][0],
                provider="seedance",
            ), task_id
        if isinstance(status, str) and status.casefold() in TERMINAL_FAILURES:
            raise AdapterFailure(
                "Seedance task failed",
                code="task_" + status.casefold(),
                request_id=task_id,
            )
        if status not in {"queued", "in_progress", "running", "processing", "pending"}:
            raise AdapterFailure(
                "Seedance returned an unknown task status",
                code="unknown_task_status",
                request_id=task_id,
            )
        time.sleep(interval)
    raise AdapterFailure(
        "Seedance task polling timed out",
        category="timeout",
        code="task_poll_timeout",
        request_id=task_id,
        retryable=True,
    )


def _run_openai(job: Mapping[str, Any]) -> tuple[Path, str | None]:
    token = _credential("OPENAI_API_KEY")
    body = compile_gpt_image_2_payload(job)
    base = _base_url("OPENAI_BASE_URL", OPENAI_BASE_URL)
    references = _reference_paths(job)
    if references:
        encoded, content_type = _multipart(body, references)
        request = urllib.request.Request(f"{base}/images/edits", data=encoded, method="POST")
        request.add_header("Authorization", f"Bearer {token}")
        request.add_header("Content-Type", content_type)
        try:
            with urllib.request.urlopen(request, timeout=300) as response:
                raw = response.read(MAX_JSON_RESPONSE + 1)
                request_id = response.headers.get("x-request-id")
        except urllib.error.HTTPError as exc:
            raise _http_failure("gpt-image-2", exc) from exc
        except TimeoutError as exc:
            raise AdapterFailure(
                "OpenAI HTTP request timed out",
                category="timeout",
                code="request_timeout",
                retryable=True,
            ) from exc
        except (urllib.error.URLError, OSError) as exc:
            raise AdapterFailure(
                "OpenAI HTTP request failed",
                category="network",
                code="network_error",
                retryable=True,
            ) from exc
        if len(raw) > MAX_JSON_RESPONSE:
            raise AdapterFailure(
                "OpenAI response is too large",
                code="response_too_large",
                request_id=request_id,
            )
        try:
            result = json.loads(raw)
        except (UnicodeError, json.JSONDecodeError) as exc:
            raise AdapterFailure(
                "OpenAI returned invalid JSON",
                code="invalid_json",
                request_id=request_id,
            ) from exc
    else:
        result, headers = _request_json(
            f"{base}/images/generations",
            provider="gpt-image-2",
            body=body,
            token=token,
        )
        request_id = headers.get("x-request-id")
    data = result.get("data") if isinstance(result, Mapping) else None
    image = data[0].get("b64_json") if isinstance(data, list) and len(data) == 1 and isinstance(data[0], Mapping) else None
    if not isinstance(image, str):
        raise AdapterFailure(
            "OpenAI did not return exactly one image",
            code="missing_image",
            request_id=request_id,
        )
    try:
        content = base64.b64decode(image, validate=True)
    except (ValueError, binascii.Error) as exc:
        raise AdapterFailure(
            "OpenAI returned invalid image data",
            code="invalid_image_data",
            request_id=request_id,
        ) from exc
    return _temporary_output(job, job["outputs"][0], content), request_id


def _run_minimax(job: Mapping[str, Any]) -> tuple[Path, str | None]:
    token = _credential("MINIMAX_API_KEY")
    body = compile_minimax_music_payload(job)
    base = _base_url("MINIMAX_BASE_URL", MINIMAX_BASE_URL)
    result, _ = _request_json(
        f"{base}/music_generation",
        provider="minimax-music",
        body=body,
        token=token,
    )
    base_resp = result.get("base_resp")
    if not isinstance(base_resp, Mapping) or base_resp.get("status_code") != 0:
        raise AdapterFailure(
            "MiniMax music generation failed",
            code=_provider_code(result) or "generation_failed",
            request_id=_safe_token(result.get("trace_id")),
        )
    data = result.get("data")
    audio = data.get("audio") if isinstance(data, Mapping) else None
    if not isinstance(data, Mapping) or data.get("status") != 2 or not isinstance(audio, str):
        raise AdapterFailure(
            "MiniMax did not return audio data",
            code="missing_audio",
            request_id=_safe_token(result.get("trace_id")),
        )
    try:
        content = bytes.fromhex(audio)
    except ValueError as exc:
        raise AdapterFailure("MiniMax returned invalid audio data") from exc
    if not content:
        raise AdapterFailure("MiniMax returned empty audio data")
    trace_id = result.get("trace_id")
    return (
        _temporary_output(job, job["outputs"][0], content),
        trace_id if isinstance(trace_id, str) else None,
    )


def _run_minimax_speech(job: Mapping[str, Any]) -> tuple[Path, str | None]:
    token = _credential("MINIMAX_API_KEY")
    body = compile_minimax_speech_payload(job)
    base = _base_url("MINIMAX_BASE_URL", MINIMAX_BASE_URL)
    result, _ = _request_json(
        f"{base}/t2a_v2",
        provider="minimax-speech",
        body=body,
        token=token,
    )
    base_resp = result.get("base_resp")
    if not isinstance(base_resp, Mapping) or base_resp.get("status_code") != 0:
        raise AdapterFailure(
            "MiniMax speech synthesis failed",
            code=_provider_code(result) or "generation_failed",
            re

#!/usr/bin/env python3
"""Offline self-test for confirmation-gated production and provider profiles."""

from __future__ import annotations

import json
import sys
import tempfile
from pathlib import Path
from typing import Any

from production_tool import (
    ConfirmationRequiredError,
    confirm_job,
    prepare_job,
    run_job,
)
from provider_adapters import (
    compile_gpt_image_2_payload,
    compile_minimax_music_payload,
    compile_seedance_payload,
)

MINIMUM_PYTHON = (3, 9)
if sys.version_info < MINIMUM_PYTHON:
    raise SystemExit("selftest.py requires Python 3.9 or newer")


def require(condition: bool, message: str) -> None:
    if not condition:
        raise RuntimeError(message)


def main() -> int:
    with tempfile.TemporaryDirectory() as directory:
        base = Path(directory)
        project = base / "project"
        project.mkdir()
        (project / "short-drama.json").write_text("{}\n", encoding="utf-8")
        job_path = project / "music-job.json"
        job: dict[str, Any] = {
            "job_id": "SELFTEST-MUSIC-001",
            "modality": "music",
            "adapter": "fixture",
            "prompt": "restrained instrumental tension",
            "references": [],
            "outputs": ["剧集/EP001/制作成果/music/cue.wav"],
            "parameters": {"is_instrumental": True},
            "overwrite": False,
        }
        job_path.write_text(json.dumps(job), encoding="utf-8")
        config = base / "adapters.json"
        fixture = Path(__file__).with_name("fixture_adapter.py")
        config.write_text(
            json.dumps(
                {
                    "adapters": {
                        "fixture": {
                            "command": [sys.executable, str(fixture)],
                            "timeout_seconds": 30,
                        }
                    }
                }
            ),
            encoding="utf-8",
        )

        preview = prepare_job(project, job_path)
        try:
            confirm_job(project, job_id=job["job_id"], confirmation="CONFIRM wrong")
        except ConfirmationRequiredError:
            pass
        else:
            raise AssertionError("a mismatched production confirmation was accepted")
        confirm_job(
            project,
            job_id=job["job_id"],
            confirmation=preview["confirmation"],
        )
        result = run_job(project, job_id=job["job_id"], adapter_config=config)
        require(result["state"] == "succeeded", "fixture production did not succeed")
        require(
            (project / job["outputs"][0]).read_bytes().startswith(b"RIFF"),
            "fixture production did not write WAV media",
        )

    require(
        compile_gpt_image_2_payload(
        {
            "modality": "image",
            "prompt": "portrait",
            "references": [],
            "outputs": ["制作成果/portrait.png"],
            "parameters": {"size": "1024x1024"},
        }
        )["model"]
        == "gpt-image-2",
        "GPT Image 2 profile compiled the wrong model",
    )
    seedance = compile_seedance_payload(
        {
            "modality": "video",
            "prompt": "slow push in",
            "references": [],
            "outputs": ["制作成果/shot.mp4"],
            "parameters": {"duration": 5, "ratio": "9:16"},
        },
        model="configured-endpoint",
        allowed_ratios={"9:16"},
        duration_range=(5, 10),
    )
    require(
        seedance.get("ratio") == "9:16" and seedance.get("duration") == 5,
        "Seedance profile did not compile explicit API parameters",
    )
    require(
        compile_minimax_music_payload(
        {
            "modality": "music",
            "prompt": "restrained score",
            "references": [],
            "outputs": ["制作成果/cue.mp3"],
            "parameters": {"is_instrumental": True},
        }
        )["model"]
        == "music-3.0",
        "MiniMax Music profile compiled the wrong model",
    )

    print("8 self-tests passed")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())