scripts/ltx_render_clips.py

"""Render LTX-2.5 clips from conditioning produced outside ComfyUI.

This is the second half of the split that `ltx_conditioning_dump.py` explains:
the encoder ran in another process and wrote a [T, 6144] tensor, and this loads
a GGUF DiT and samples from that tensor. Nothing here knows or cares which
encoder produced it, which is the point - a compressed encoder ComfyUI cannot
load still drives the same DiT, and two builds can be compared frame for frame
because everything downstream of the conditioning is identical.

Why a script instead of a ComfyUI graph: the injection point does not exist as a
node. `CLIPTextEncode` is the only producer of LTX conditioning and it needs a
loadable encoder. Driving ComfyUI's classes directly is a smaller and more
honest change than a custom node that fakes a CLIP object.

Three things are deliberately not the vendor default:

* `Guider_Basic` rather than dual CFG. The published workflow sets `video_cfg`
  and `audio_cfg` to 1.0, and at 1.0 `Guider_DualCFG.predict_noise` drops both
  the negative and middle conds and returns the positive prediction unchanged.
  The two are the same computation; this one does not need a negative prompt.
* no latent upscale pass. The vendor graph samples at base resolution, upscales
  the latent 2x and samples again, and this script stops after the first pass.
  Both builds stop identically, so it never affected a comparison.

  **The reason given here for stopping was wrong.** It claimed the second pass,
  at 4x the token count, is what a 16 GB card cannot afford. Measured: 233
  frames from 512x320 to 1024x640, both passes plus the decode, peaks at
  10.03 GiB and finishes in 312 s on a V100. It fits, and it is the difference
  between soft and sharp - 4.1x the Laplacian variance on the same pixel grid.
  See `reports/ltx25_av_alignment/two_pass/`. `ltx_av_guide.upscale_av_latent`
  performs the pass; this script has not been rewired to use it.
* tiled VAE decode by default. A 97-frame decode at full size peaks well above
  what is left after a 10.6 GB DiT.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import subprocess
import sys
import time
import wave
from fractions import Fraction
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))

from pacific.ltx_av import (
    generation_manifest_identity,
    target_sample_count,
    validate_generation_manifest,
)
from pacific.strict_json import strict_json_loads

#: Overridable because a Space, a rented box and this workstation each put
#: ComfyUI somewhere different, and the import below needs the real path.
COMFY = Path(os.environ.get("LTX_COMFY_DIR", "/home/topabaem/ComfyUI"))

#: The distilled schedule from Lightricks' own graph: nine sigmas, eight steps.
#: Not a guess and not tuned here - a different schedule would be a second
#: variable between builds.
DISTILLED_SIGMAS = "1.0,0.99375,0.9875,0.98125,0.975,0.909375,0.725,0.421875,0.0"


def import_comfy(comfy_args: list[str] | None = None):
    """Import ComfyUI with a substituted `sys.argv`.

    `comfy.cli_args` runs argparse at import time against the real `sys.argv`,
    so this script's own flags would be parsed as ComfyUI's and abort. Swapping
    argv is the whole trick, and it is also the only way to set ComfyUI's memory
    policy - there is no API for it, `--reserve-vram` is read once at import.

    Memory policy matters on a card this size. The card is shared: an unrelated
    `llama-server` holds 1.66 GB and is not ours to kill. Left to itself ComfyUI
    loads all 10.6 GB of DiT resident and then has nothing left for the two
    embeddings connectors or the dequantisation buffers, and dies inside the
    Q3_K bit unpack. `--reserve-vram` alone did not change that - it trims the
    budget without changing the loading mode - so `--lowvram` is the flag that
    actually makes it stream module by module.
    """
    saved = sys.argv[:]
    sys.argv = [saved[0], *(comfy_args or [])]
    sys.path.insert(0, str(COMFY))
    try:
        # Without this `cli_args` runs `parse_args([])` and every flag below is
        # silently the default. `main.py` is the only caller that switches it on,
        # so importing ComfyUI as a library gets defaults and no warning - two
        # runs with different flags produce byte-identical OOM messages, which is
        # how this was found.
        import comfy.options
        comfy.options.enable_args_parsing()

        import comfy.model_management
        import comfy.sample
        import comfy.samplers
        import comfy.sd
        import comfy.utils
        import folder_paths  # noqa: F401
    finally:
        sys.argv = saved


def unwrap(node_output):
    """ComfyUI nodes return `io.NodeOutput`; the values live behind it."""
    for attr in ("result", "_result", "values"):
        if hasattr(node_output, attr):
            value = getattr(node_output, attr)
            if value is not None:
                return value
    return node_output


def load_gguf_unet(path: Path):
    """Load the DiT through ComfyUI-GGUF.

    Its dequantisation is written in plain torch, which is the only reason any
    of this runs on a V100: every ComfyUI-native quantised format - fp8, nvfp4,
    mxfp8 - dispatches to tensor-core kernels that need compute capability 8.9.
    """
    import importlib

    sys.path.insert(0, str(COMFY / "custom_nodes"))
    module = importlib.import_module("ComfyUI-GGUF")
    loader = module.NODE_CLASS_MAPPINGS["UnetLoaderGGUF"]()

    import folder_paths
    folder_paths.add_model_folder_path("unet_gguf", str(path.parent))
    return unwrap(loader.load_unet(path.name))[0]


def apply_lora(model, path: Path, strength: float):
    """Patch a LoRA onto the GGUF DiT.

    Lightricks' own `dev` workflow never runs `dev` bare - it loads
    `ltx-2.5-22b-distilled-lora-450` on top at strength 0.5. Run without it, this
    checkpoint produces soft, under-resolved frames at every schedule, step count
    and precision tried, while the distilled transformer is sharp in eight steps.
    ComfyUI-GGUF supports this: `GGUFModelPatcher.patch_weight_to_device` attaches
    patches to the quantized tensor rather than materialising it.
    """
    import comfy.sd
    import comfy.utils
    import folder_paths
    folder_paths.add_model_folder_path("loras", str(path.parent))
    lora = comfy.utils.load_torch_file(str(path), safe_load=True)
    patched, _ = comfy.sd.load_lora_for_models(model, None, lora, strength, 0)
    return patched


def load_video_vae(path: Path):
    """The metadata is not optional here.

    `comfy.sd.VAE` reads the LTX decoder's channel widths out of
    `metadata["config"]` (sd.py:624). Without it the state dict still matches the
    LTX branch on tensor names, so detection succeeds and then loads a
    differently sized LTX VAE - the failure is 200 lines of shape mismatch, not a
    clear "wrong file".
    """
    import comfy.sd
    import comfy.utils
    sd, metadata = comfy.utils.load_torch_file(str(path), return_metadata=True)
    return comfy.sd.VAE(sd=sd, metadata=metadata)


def load_audio_vae(path: Path):
    import comfy.utils
    from comfy.ldm.lightricks.vae.audio_vae import AudioVAE
    sd, metadata = comfy.utils.load_torch_file(str(path), return_metadata=True)
    return AudioVAE(sd, metadata)


def make_latents(width: int, height: int, length: int, fps: Fraction, audio_vae):
    """Empty video and audio latents, joined into the nested pair the DiT wants."""
    import comfy.model_management
    import comfy.nested_tensor
    import torch
    from comfy_extras.nodes_lt_audio import LTXVEmptyLatentAudio

    video = torch.zeros(
        [1, 128, ((length - 1) // 8) + 1, height // 32, width // 32],
        device=comfy.model_management.intermediate_device())
    audio = unwrap(LTXVEmptyLatentAudio.execute(
        frames_number=length, frame_rate=float(fps), batch_size=1, audio_vae=audio_vae))
    if isinstance(audio, (tuple, list)):
        audio = audio[0]

    latent = {}
    latent.update({"samples": video})
    latent.update({k: v for k, v in audio.items() if k != "samples"})
    latent["samples"] = comfy.nested_tensor.NestedTensor((video, audio["samples"]))
    return latent


def wrap_conditioning(tensor, fps: Fraction):
    """A [T, 6144] tensor as a ComfyUI CONDITIONING, with the flag it needs.

    `unprocessed_ltxav_embeds` is not a hint, it selects a different graph. The
    aggregate output is 6144 wide, which is exactly `cross_attention_dim +
    audio_cross_attention_dim`, so without the flag `preprocess_text_embeds`
    matches its early return and hands the raw aggregates straight to
    cross-attention - skipping both embeddings connectors. It renders, it just
    renders mud. `LTXAVTEModel` sets this flag on every dual_linear encode; we
    produce the same tensor, so we owe it too.
    """
    import node_helpers
    cond = [[tensor.unsqueeze(0), {}]]
    return node_helpers.conditioning_set_values(cond, {
        "frame_rate": float(fps),
        "unprocessed_ltxav_embeds": True,
    })


def build_sigmas(model, sigmas_text: str, steps: int, scheduler: str,
                 latent=None, max_shift: float = 2.05, base_shift: float = 0.95,
                 terminal: float = 0.1):
    """The literal vendor schedule, `LTXVScheduler`, or a generic ComfyUI one.

    The nine hardcoded sigmas belong to the distilled transformer and only to
    it - a schedule solved for a model trained to land in eight steps.

    For `dev`, ComfyUI's generic schedulers are the wrong tool and it shows on
    screen. `normal` at 30 steps spends 27 of them between 1.0 and 0.28 and then
    drops 0.28 -> 0.011 -> 0, so the low-sigma range where structure resolves is
    barely sampled at all and the render comes out hazy and washed out.

    `LTXVScheduler` is what Lightricks' own dev workflow uses. It builds a
    linspace, applies a shift interpolated from the *token count* of the latent -
    which is why `latent` is worth passing - and then stretches the tail so the
    last non-zero sigma lands on `terminal` rather than wherever the curve fell.
    Defaults here are that workflow's: 2.05 / 0.95 / stretch / 0.1.
    """
    import comfy.samplers
    import torch
    if steps <= 0:
        return torch.FloatTensor([float(s) for s in sigmas_text.split(",")])
    if scheduler == "ltxv":
        from comfy_extras.nodes_lt import LTXVScheduler
        out = unwrap(LTXVScheduler.execute(
            steps=steps, max_shift=max_shift, base_shift=base_shift,
            stretch=True, terminal=terminal, latent=latent))
        if isinstance(out, (tuple, list)):
            out = out[0]
        return out.cpu()
    return comfy.samplers.calculate_sigmas(
        model.get_model_object("model_sampling"), scheduler, steps).cpu()


def sample(model, conditioning, latent, sigmas, seed: int, fps: Fraction,
           cfg: float = 1.0, negative=None, sampler_name: str = "euler_ancestral"):
    """Sample one clip. `cfg` above 1 needs `negative` and buys prompt adherence.

    The vendor graph sets both CFG scales to 1.0, which makes the guider return
    the conditional prediction unchanged - correct for a distilled model, which
    is trained with the guidance already baked in, and half the compute. On the
    distilled transformer raising CFG changed nothing at all, up to 6.0 with an
    explicit anti-human negative, which is consistent with that: there is no
    guidance direction left to scale. On `dev` it should be the main knob.
    """
    import comfy.model_management
    import comfy.sample
    import comfy.samplers
    from comfy_extras.nodes_custom_sampler import Guider_Basic, Noise_RandomNoise

    cond = wrap_conditioning(conditioning, fps)

    if cfg > 1.0:
        if negative is None:
            raise SystemExit("--cfg above 1 needs a negative conditioning index")
        guider = comfy.samplers.CFGGuider(model)
        guider.set_conds(cond, wrap_conditioning(negative, fps))
        guider.set_cfg(cfg)
    else:
        guider = Guider_Basic(model)
        guider.set_conds(cond)

    sampler = comfy.samplers.sampler_object(sampler_name)
    noise = Noise_RandomNoise(seed)

    samples = latent["samples"]
    samples = comfy.sample.fix_empty_latent_channels(guider.model_patcher, samples)
    out = guider.sample(noise.generate_noise({"samples": samples}), samples,
                        sampler, sigmas, denoise_mask=None, callback=None,
                        disable_pbar=False, seed=seed)
    return out.to(comfy.model_management.intermediate_device())


def decode(samples, video_vae, audio_vae, tile: int, temporal: int):
    """Inference mode is not an optimisation here, it is required.

    `VAE.process_output` finishes with `image.add_(1.0).div_(2.0)`, an in-place
    write. ComfyUI's execution engine runs every node inside `inference_mode`,
    so the tensor it writes to was created there too and the write is legal.
    Called from an ordinary script the tiles come back as inference tensors and
    the same line raises. Wrapping the decode puts us back on ComfyUI's terms.
    """
    import comfy.model_management
    import torch

    video_latent, audio_latent = samples.unbind()
    # The device moves stay OUTSIDE inference mode, and that is not tidiness.
    # `Module.to` rebinds every parameter, and a parameter rebound inside
    # inference mode is an inference tensor for the rest of the process. Later,
    # a `VAE.encode` for an image guide hits `torch.cudnn_convolution` with an
    # inference weight and dies with "Inference tensors cannot be saved for
    # backward" - in a different function, on a different request, with nothing
    # to point back here. Volta hides it, because its fp32 cast skips the branch
    # that reaches that op at all.
    video_vae.first_stage_model.to(comfy.model_management.get_torch_device())
    with torch.inference_mode():
        frames = video_vae.decode_tiled(video_latent, tile_x=tile, tile_y=tile,
                                        overlap=tile // 8, tile_t=temporal,
                                        overlap_t=temporal // 8)
    video_vae.first_stage_model.to("cpu")
    comfy.model_management.soft_empty_cache()

    with torch.inference_mode():
        audio = audio_vae.decode(audio_latent)
    return frames.clone(), audio.clone(), int(audio_vae.output_sample_rate)


def write_clip(
    frames,
    audio,
    sample_rate: int,
    fps: Fraction,
    out: Path,
    raw_video: Path | None = None,
) -> tuple[Path, int]:
    """Pipe raw frames into ffmpeg rather than leaving a PNG sequence behind."""
    import numpy as np
    import torch

    array = frames
    if array.dim() == 5:
        array = array[0]
    array = array.float().clamp(0, 1).mul(255).round().to("cpu", torch.uint8).numpy()
    if array.shape[-1] != 3:
        array = np.moveaxis(array, 1, -1)
    count, height, width, _ = array.shape
    target_samples = target_sample_count(count, fps, sample_rate)
    if raw_video is not None:
        raw_video.parent.mkdir(parents=True, exist_ok=True)
        np.save(raw_video, array)

    wav = out.with_suffix(".wav")
    waveform = audio
    if waveform.dim() == 3:
        waveform = waveform[0]
    pcm = waveform.transpose(0, 1).float().clamp(-1, 1).mul(32767).to("cpu", torch.int16).numpy()
    with wave.open(str(wav), "wb") as f:
        f.setnchannels(pcm.shape[1])
        f.setsampwidth(2)
        f.setframerate(sample_rate)
        f.writeframes(pcm.tobytes())

    command = [
        "ffmpeg", "-y", "-loglevel", "error",
        "-f", "rawvideo", "-pix_fmt", "rgb24",
        "-s", f"{width}x{height}", "-r", str(fps), "-i", "-",
        "-i", str(wav),
        "-c:v", "libx264", "-preset", "medium", "-crf", "17", "-pix_fmt", "yuv420p",
        "-af", f"apad,atrim=end_sample={target_samples}",
        "-c:a", "aac", "-b:a", "192k", str(out),
    ]
    process = subprocess.Popen(command, stdin=subprocess.PIPE)
    process.communicate(array.tobytes())
    if process.returncode != 0:
        raise SystemExit(f"ffmpeg failed for {out}")
    print(f"  wrote {out}  {count} frames {width}x{height}", flush=True)
    return wav, int(pcm.shape[0])


def sha256_path(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


def artifact(path: Path) -> dict[str, str]:
    return {"path": str(path.resolve()), "sha256": sha256_path(path)}


def append_manifest(path: Path, record: dict) -> None:
    validate_generation_manifest(record)
    existing: dict[str, dict] = {}
    if path.exists():
        for line in path.read_text(encoding="utf-8").splitlines():
            if line.strip():
                row = strict_json_loads(line)
                existing[row["sample_id"]] = row
    previous = existing.get(record["sample_id"])
    if previous is not None:
        if generation_manifest_identity(previous) != generation_manifest_identity(record):
            raise SystemExit(f"manifest already has a different {record['sample_id']}")
        return
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8") as handle:
        handle.write(json.dumps(record, allow_nan=False, sort_keys=True) + "\n")


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--conditioning", type=Path, required=True)
    parser.add_argument("--gguf", type=Path, required=True)
    parser.add_argument("--video-vae", type=Path, required=True)
    parser.add_argument("--audio-vae", type=Path, required=True)
    parser.add_argument("--out-dir", type=Path, required=True)
    parser.add_argument("--width", type=int, default=768)
    parser.add_argument("--height", type=int, default=512)
    parser.add_argument("--lengths", default="73,49,97,49,73",
                        help="frames per clip; LTX needs 8n+1")
    parser.add_argument("--fps", type=Fraction, default=Fraction(24, 1))
    parser.add_argument("--seed", type=int, default=20260813)
    parser.add_argument("--sigmas", default=DISTILLED_SIGMAS,
                        help="literal schedule; ignored when --steps is set")
    parser.add_argument("--steps", type=int, default=0,
                        help="compute the schedule instead; use this for `dev`")
    parser.add_argument("--scheduler", default="ltxv",
                        help="'ltxv' for LTXVScheduler, or a ComfyUI scheduler name")
    parser.add_argument("--max-shift", type=float, default=2.05)
    parser.add_argument("--base-shift", type=float, default=0.95)
    parser.add_argument("--terminal", type=float, default=0.1)
    parser.add_argument("--sampler", default="euler_ancestral",
                        help="euler is the usual choice for a non-distilled model")
    parser.add_argument("--tile", type=int, default=512)
    parser.add_argument("--temporal", type=int, default=32)
    parser.add_argument("--only", type=int, default=None,
                        help="render a single prompt index; for the smoke test")
    parser.add_argument("--limit", type=int,
                        help="render only the first N prompts; for diagnostic subsets")
    parser.add_argument("--cfg", type=float, default=1.0,
                        help="classifier-free guidance; >1 doubles cost per step")
    parser.add_argument("--negative-index", type=int, default=None,
                        help="prompt index in the same file to use as the "
                             "unconditional branch when --cfg is above 1")
    parser.add_argument("--lora", type=Path, default=None,
                        help="LoRA to patch onto the DiT; the vendor dev "
                             "workflow uses the distilled LoRA at 0.5")
    parser.add_argument("--lora-strength", type=float, default=0.5)
    parser.add_argument("--fp32", action="store_true",
                        help="skip --fp16-unet; ComfyUI then casts to fp32. "
                             "Slower and twice the activation memory, but fp16 "
                             "saturation on Volta shows up as hazy low-contrast "
                             "output rather than as an error")
    parser.add_argument("--reserve-vram", type=float, default=1.0,
                        help="GB ComfyUI must leave free")
    parser.add_argument("--vram-mode", choices=("normal", "lowvram", "novram"),
                        default="normal",
                        help="how much of the DiT ComfyUI may keep resident")
    parser.add_argument("--stage-dir", type=Path,
                        help="save generated latents and raw decoder frames")
    parser.add_argument("--manifest", type=Path,
                        help="append a deterministic stage manifest; requires --stage-dir")
    parser.add_argument("--profile-id", default="")
    parser.add_argument("--model-kind", choices=("baseline", "compressed", "ablation"),
                        default="compressed")
    args = parser.parse_args()
    if args.manifest is not None and args.stage_dir is None:
        parser.error("--manifest requires --stage-dir")
    if args.only is not None and args.limit is not None:
        parser.error("--only and --limit are mutually exclusive")
    if args.limit is not None and args.limit <= 0:
        parser.error("--limit must be positive")

    # Volta has fp16 tensor cores and no bf16 ones, so `should_use_bf16` is
    # False - but the checkpoint declares bf16, and ComfyUI resolves that
    # mismatch by setting `manual_cast_dtype` to fp32. Every matmul then runs at
    # fp32: double the activation memory and half the throughput, which is the
    # difference between this fitting and not. Forcing fp16 removes the cast
    # entirely (`manual_cast` becomes None).
    comfy_args = ["--reserve-vram", str(args.reserve_vram)]
    if not args.fp32:
        comfy_args.insert(0, "--fp16-unet")
    if args.vram_mode != "normal":
        comfy_args.append(f"--{args.vram_mode}")
    import_comfy(comfy_args)
    import torch

    bundle = torch.load(args.conditioning, weights_only=False)
    prompts = bundle["prompts"]
    prompt_ids = bundle.get("prompt_ids", [f"prompt-{i:02d}" for i in range(len(prompts))])
    conditioning = bundle["conditioning"]
    label = bundle.get("label", args.conditioning.stem)
    profile_id = args.profile_id or label
    print(f"{len(prompts)} prompts from build '{label}'", flush=True)

    lengths = [int(v) for v in args.lengths.split(",")]
    if len(lengths) < len(prompts):
        lengths = lengths + [lengths[-1]] * (len(prompts) - len(lengths))

    if args.only is not None:
        indices = [args.only]
    else:
        count = len(prompts) if args.limit is None else min(args.limit, len(prompts))
        indices = list(range(count))
    negative = None
    if args.cfg > 1.0:
        if args.negative_index is None:
            raise SystemExit("--cfg above 1 requires --negative-index")
        negative = conditioning[args.negative_index]
        indices = [i for i in indices if i != args.negative_index]
        print(f"cfg {args.cfg} against prompt [{args.negative_index}] as negative",
              flush=True)

    args.out_dir.mkdir(parents=True, exist_ok=True)
    audio_vae = load_audio_vae(args.audio_vae)
    video_vae = load_video_vae(args.video_vae)
    model = load_gguf_unet(args.gguf)
    if args.lora is not None:
        model = apply_lora(model, args.lora, args.lora_strength)
        print(f"lora {args.lora.name} @ {args.lora_strength}", flush=True)
    print(f"models loaded; scheduler {args.scheduler}, sampler {args.sampler}, "
          f"cfg {args.cfg}", flush=True)

    for i in indices:
        clip_started = time.perf_counter()
        if torch.cuda.is_available():
            torch.cuda.reset_peak_memory_stats()
        length = lengths[i]
        seed = args.seed + i
        sample_id = f"{profile_id}__{prompt_ids[i]}__seed-{seed}"
        print(f"[{i}] {length} frames  {prompts[i][:70]}", flush=True)
        latent = make_latents(args.width, args.height, length, args.fps, audio_vae)
        # LTXVScheduler shifts by token count, so the schedule is per clip.
        sigmas = build_sigmas(model, args.sigmas, args.steps, args.scheduler,
                              latent=latent, max_shift=args.max_shift,
                              base_shift=args.base_shift, terminal=args.terminal)
        print(f"    {len(sigmas) - 1} steps  "
              f"{[round(float(x), 3) for x in sigmas[:4]]} .. "
              f"{[round(float(x), 3) for x in sigmas[-3:]]}", flush=True)
        sample_started = time.perf_counter()
        samples = sample(
            model,
            conditioning[i],
            latent,
            sigmas,
            seed,
            args.fps,
            cfg=args.cfg,
            negative=negative,
            sampler_name=args.sampler,
        )
        sample_seconds = time.perf_counter() - sample_started
        latent_path = None
        raw_video_path = None
        if args.stage_dir is not None:
            latent_path = args.stage_dir / "latent" / f"{sample_id}.pt"
            raw_video_path = args.stage_dir / "raw" / f"{sample_id}-frames.npy"
            latent_path.parent.mkdir(parents=True, exist_ok=True)
            video_latent, audio_latent = samples.unbind()
            torch.save(
                {"video": video_latent.cpu(), "audio": audio_latent.cpu()},
                latent_path,
            )
        decode_started = time.perf_counter()
        frames, audio, rate = decode(
            samples, video_vae, audio_vae, args.tile, args.temporal
        )
        decode_seconds = time.perf_counter() - decode_started
        output_name = f"{sample_id}.mp4" if "prompt_ids" in bundle else f"{label}-{i:02d}.mp4"
        output = args.out_dir / output_name
        mux_started = time.perf_counter()
        wav, audio_samples = write_clip(
            frames, audio, rate, args.fps, output, raw_video=raw_video_path
        )
        mux_seconds = time.perf_counter() - mux_started
        total_clip_seconds = time.perf_counter() - clip_started
        peak_allocated = (
            int(torch.cuda.max_memory_allocated()) if torch.cuda.is_available() else 0
        )
        if args.manifest is not None:
            from ltx_av_timing_audit import audit

            timing = audit(output)
            record = {
                "schema_version": 1,
                "sample_id": sample_id,
                "model_kind": args.model_kind,
                "profile": profile_id,
                "checkpoint": bundle.get("checkpoint", ""),
                "checkpoint_sha256": bundle.get("checkpoint_sha256", ""),
                "profile_hash": bundle.get("profile_hash", ""),
                "conditioning_sha256": sha256_path(args.conditioning),
                "prompt_id": prompt_ids[i],
                "prompt": prompts[i],
                "seed": seed,
                "frame_count": timing["video_frames"],
                "fps": timing["fps"],
                "audio_sample_count": audio_samples,
                "audio_sample_rate": rate,
                "first_video_pts": timing["video_first_pts"],
                "last_video_end_pts": timing["video_last_end_pts"],
                "video_time_base": timing["video_time_base"],
                "first_audio_pts": timing["audio_first_packet_pts"],
                "last_audio_end_pts": timing["audio_last_end_pts"],
                "audio_time_base": timing["audio_time_base"],
                "sample_seconds": sample_seconds,
                "decode_seconds": decode_seconds,
                "mux_seconds": mux_seconds,
                "total_clip_seconds": total_clip_seconds,
                "torch_peak_allocated_bytes": peak_allocated,
                "generation_command": [sys.executable, *sys.argv],
                "artifacts": {
                    "latent": artifact(latent_path),
                    "raw_video": artifact(raw_video_path),
                    "raw_audio": artifact(wav),
                    "muxed": artifact(output),
                },
            }
            append_manifest(args.manifest, record)
    return 0


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