ltx/runtime.py

from __future__ import annotations

import gc
import json
import time
from dataclasses import dataclass
from typing import Callable

import torch
from huggingface_hub import snapshot_download
from diffusers import (
    FlowMatchEulerDiscreteScheduler,
    LTX2ConditionPipeline,
    LTX2ImageToVideoPipeline,
    LTX2InContextPipeline,
    LTX2LatentUpsamplePipeline,
    LTX2VideoDiffusionDecodePipeline,
    LTX2VideoDiffusionDecoderModel,
)
from diffusers.models.autoencoders.ltx2_diffusion_decoder import LTX2VideoVaeNeighborhoodNattenProcessor
from diffusers.pipelines.ltx2.latent_upsampler import LTX2LatentUpsamplerModel

from ltx import model_loading


LogFn = Callable[[str], None]


@dataclass(frozen=True)
class RuntimeBuildPolicy:
    allow_patterns: tuple[str, ...]
    runtime_profile: str
    is_full_sft_profile: bool
    auto_duration_enabled: bool
    auto_duration_min_seconds: float
    auto_duration_max_seconds: float
    diffusion_decoder_enabled: bool
    prompt_enhancer_enabled: bool
    ic_colorizer_enabled: bool
    ic_pixel_upscaler_enabled: bool
    ic_inoutpaint_enabled: bool
    transformer_override_repo_id: str | None
    transformer_override_path: str | None
    transformer_override_revision: str | None
    text_encoder_override_repo_id: str | None
    text_encoder_override_path: str | None
    text_encoder_override_revision: str | None
    prompt_enhancer_repo_id: str
    prompt_enhancer_revision: str | None
    prompt_enhancer_policy: str
    full_sft_transformer_path: str
    full_sft_transformer_repo: str
    full_sft_transformer_revision: str | None
    full_sft_stage2_lora_repo: str
    full_sft_stage2_lora_revision: str | None
    full_sft_stage2_lora_weight_name: str
    full_sft_stage2_lora_strength: float
    full_sft_stage2_adapter_name: str
    attention_backend: str


@dataclass(frozen=True)
class RuntimePreloadPolicy:
    candidate_id: str
    model_repo_id: str
    model_revision: str | None
    model_quantization_policy_requested: str
    quantization_policy: str
    model_runtime_profile_requested: str
    runtime_profile: str
    config_warnings: tuple[str, ...]
    worker_uuid: str
    canonical_model_repo_id: str
    canonical_model_revision: str | None
    is_full_sft_profile: bool
    is_zerogpu: bool
    auto_duration_min_seconds: float
    auto_duration_max_seconds: float


@dataclass(frozen=True)
class RuntimeArtifacts:
    pipe: object
    pipe_i2v: object
    pipe_condition: object
    pipe_ic: object | None
    upsample_pipe: object
    diffusion_decode_pipe: object | None
    prompt_enhancer_model: object | None
    prompt_enhancer_processor: object | None
    model_sources: dict
    attention_state: dict
    phases: dict
    model_dir: str


@dataclass(frozen=True)
class RuntimePreloadResult:
    runtime: RuntimeArtifacts | None
    state: dict


def _override_requested(repo_id) -> bool:
    return bool(str(repo_id or "").strip())


def _source_record(requested: dict | None = None) -> dict:
    return {"requested": requested, "effective": None, "fallback": False, "fallback_reason": None}


def _mark_component_fallback(record: dict, reason: str, log_fn: LogFn) -> None:
    record["fallback"] = True
    record["fallback_reason"] = reason
    log_fn(f"[MODEL_OVERRIDE] fallback component reason={reason}")


def attempt_runtime_for_base(
    repo_id: str,
    revision: str | None,
    base_record: dict,
    policy: str,
    *,
    token: str | None,
    build_policy: RuntimeBuildPolicy,
    log_fn: LogFn,
) -> tuple:
    # Phase timings remain explicit because startup residency is part of the product evidence.
    phases = {}
    attempt_started = time.monotonic()

    phase_started = time.monotonic()
    log_fn(f"[D1] downloading trimmed base snapshot repo_id={repo_id} revision={revision!r}")
    model_dir = snapshot_download(
        repo_id=repo_id,
        revision=revision or None,
        token=token,
        allow_patterns=build_policy.allow_patterns,
        max_workers=8,
    )
    phases["snapshot_download_seconds"] = time.monotonic() - phase_started
    base_record["effective"] = {
        "repo_id": repo_id,
        "revision": revision,
        "quantization_policy": policy,
        "runtime_profile": build_policy.runtime_profile,
    }

    # Requested/effective source records are kept separate so fallback is never silent.
    model_sources = {
        "base": base_record,
        "transformer": _source_record(
            {
                "repo_id": str(build_policy.transformer_override_repo_id or "").strip() or None,
                "path": str(build_policy.transformer_override_path or "").strip() or None,
                "revision": str(build_policy.transformer_override_revision or "").strip() or None,
            }
            if _override_requested(build_policy.transformer_override_repo_id)
            else None
        ),
        "text_encoder": _source_record(
            {
                "repo_id": str(build_policy.text_encoder_override_repo_id or "").strip() or None,
                "path": str(build_policy.text_encoder_override_path or "").strip() or None,
                "revision": str(build_policy.text_encoder_override_revision or "").strip() or None,
            }
            if _override_requested(build_policy.text_encoder_override_repo_id)
            else None
        ),
        "prompt_enhancer": _source_record(
            {
                "enabled": bool(build_policy.prompt_enhancer_enabled),
                "repo_id": str(build_policy.prompt_enhancer_repo_id or "").strip() or None,
                "revision": str(build_policy.prompt_enhancer_revision or "").strip() or None,
                "quantization_policy": build_policy.prompt_enhancer_policy,
            }
        ),
        "duration_head": _source_record(
            {
                "enabled": bool(build_policy.auto_duration_enabled),
                "repo_id": repo_id,
                "path": "duration_head",
                "revision": revision,
            }
        ),
    }

    # Prepare optional/full-profile component overrides before constructing the base pipeline.
    phase_started = time.monotonic()
    if build_policy.is_full_sft_profile:
        if _override_requested(build_policy.transformer_override_repo_id):
            raise RuntimeError(
                "Use FULL_SFT_TRANSFORMER_REPO_ID / REVISION / PATH for the full_sft_nf4 profile; "
                "the generic TRANSFORMER_OVERRIDE_* surface is reserved for distilled_nf4."
            )
        transformer_override = model_loading.load_full_sft_transformer(
            model_dir,
            model_sources["transformer"],
            policy,
            repo_id,
            revision,
            path=build_policy.full_sft_transformer_path,
            source_repo=build_policy.full_sft_transformer_repo,
            source_revision=build_policy.full_sft_transformer_revision,
            token=token,
            runtime_profile=build_policy.runtime_profile,
        )
    else:
        transformer_override = model_loading.load_transformer_override(
            model_dir,
            model_sources["transformer"],
            policy,
            repo_id=build_policy.transformer_override_repo_id,
            path=build_policy.transformer_override_path,
            revision=build_policy.transformer_override_revision,
            token=token,
            log_fn=log_fn,
            mark_fallback=lambda record, reason: _mark_component_fallback(record, reason, log_fn),
        )
    text_encoder_override = model_loading.load_text_encoder_override(
        model_sources["text_encoder"],
        policy,
        repo_id=build_policy.text_encoder_override_repo_id,
        path=build_policy.text_encoder_override_path,
        revision=build_policy.text_encoder_override_revision,
        token=token,
        log_fn=log_fn,
        mark_fallback=lambda record, reason: _mark_component_fallback(record, reason, log_fn),
    )
    phases["override_prepare_seconds"] = time.monotonic() - phase_started

    phase_started = time.monotonic()
    built_pipeline, pipeline_build_error = model_loading.try_build_base_pipeline(
        model_dir,
        transformer_override=transformer_override,
        text_encoder_override=text_encoder_override,
        policy=policy,
        auto_duration_enabled=build_policy.auto_duration_enabled,
    )
    if built_pipeline is None and text_encoder_override is not None:
        reason = f"pipeline integration failed: {pipeline_build_error}"
        log_fn(f"[MODEL_OVERRIDE] text_encoder integration FAILED: {reason}")
        _mark_component_fallback(model_sources["text_encoder"], reason, log_fn)
        text_encoder_override = None
        gc.collect()
        built_pipeline, pipeline_build_error = model_loading.try_build_base_pipeline(
            model_dir,
            transformer_override=transformer_override,
            text_encoder_override=None,
            policy=policy,
            auto_duration_enabled=build_policy.auto_duration_enabled,
        )

    if built_pipeline is None and transformer_override is not None:
        reason = f"pipeline integration failed after prior fallback: {pipeline_build_error}"
        if build_policy.is_full_sft_profile:
            raise RuntimeError(f"Full/SFT transformer pipeline integration failed; distilled fallback forbidden: {reason}")
        log_fn(f"[MODEL_OVERRIDE] transformer integration FAILED: {reason}")
        _mark_component_fallback(model_sources["transformer"], reason, log_fn)
        transformer_override = None
        gc.collect()
        built_pipeline, pipeline_build_error = model_loading.try_build_base_pipeline(
            model_dir,
            transformer_override=None,
            text_encoder_override=None,
            policy=policy,
            auto_duration_enabled=build_policy.auto_duration_enabled,
        )

    if built_pipeline is None:
        raise RuntimeError(f"base pipeline construction failed: {pipeline_build_error}")
    pipe, base_quantization = built_pipeline
    phases["pipeline_load_seconds"] = time.monotonic() - phase_started

    if transformer_override is None:
        model_sources["transformer"]["effective"] = {
            "kind": "base_component",
            "repo_id": repo_id,
            "path": "transformer",
            "revision": revision,
            "quantization": base_quantization.get("transformer"),
        }
    if text_encoder_override is None:
        model_sources["text_encoder"]["effective"] = {
            "kind": "base_component",
            "repo_id": repo_id,
            "path": "text_encoder",
            "revision": revision,
            "quantization": base_quantization.get("text_encoder"),
        }

    pipe.vae.enable_tiling()

    if build_policy.is_full_sft_profile:
        pipe.scheduler = FlowMatchEulerDiscreteScheduler.from_config(
            pipe.scheduler.config, use_dynamic_shifting=True, shift_terminal=0.1
        )

    duration_record = model_sources["duration_head"]
    if build_policy.auto_duration_enabled and getattr(pipe, "duration_head", None) is not None:
        duration_record["effective"] = {
            "kind": "base_component",
            "repo_id": repo_id,
            "path": "duration_head",
            "revision": revision,
            "dtype": str(getattr(pipe.duration_head, "dtype", torch.bfloat16)).replace("torch.", ""),
            "residency": "module-scope CUDA packed; small component",
            "bounds_seconds": [build_policy.auto_duration_min_seconds, build_policy.auto_duration_max_seconds],
        }
    elif build_policy.auto_duration_enabled:
        duration_record["fallback"] = True
        duration_record["fallback_reason"] = "duration_head component unavailable in loaded pipeline"
        duration_record["effective"] = {"kind": "disabled", "reason": duration_record["fallback_reason"]}
    else:
        duration_record["effective"] = {"kind": "disabled", "reason": "disabled by space_config.py"}

    phase_started = time.monotonic()
    latent_upsampler = LTX2LatentUpsamplerModel.from_pretrained(
        model_dir,
        subfolder="latent_upsampler",
        dtype=torch.bfloat16,
    )
    phases["latent_upsampler_load_seconds"] = time.monotonic() - phase_started

    stage2_local_path = None
    resolved_stage2_revision = None
    if build_policy.is_full_sft_profile:
        phase_started = time.monotonic()
        stage2_local_path, resolved_stage2_revision = model_loading.prepare_full_sft_stage2_lora(
            model_dir,
            repo_id,
            revision,
            repo_id=build_policy.full_sft_stage2_lora_repo,
            revision=build_policy.full_sft_stage2_lora_revision,
            weight_name=build_policy.full_sft_stage2_lora_weight_name,
            token=token,
        )
        phases["full_sft_stage2_lora_hub_prepare_seconds"] = time.monotonic() - phase_started

    phase_started = time.monotonic()
    pipe.to("cuda")
    latent_upsampler.to("cuda")
    attention_state = model_loading.maybe_enable_attention_backend(pipe, build_policy.attention_backend)
    phases["cuda_pack_and_attention_seconds"] = time.monotonic() - phase_started

    stage2_record = _source_record(None)
    stage2_record["requested"] = {
        "enabled": bool(build_policy.is_full_sft_profile),
        "repo_id": build_policy.full_sft_stage2_lora_repo,
        "weight_name": build_policy.full_sft_stage2_lora_weight_name,
        "revision": build_policy.full_sft_stage2_lora_revision,
        "strength": build_policy.full_sft_stage2_lora_strength,
    }
    if build_policy.is_full_sft_profile:
        phase_started = time.monotonic()
        weight_name = stage2_record["requested"]["weight_name"]
        try:
            if stage2_local_path is None:
                raise RuntimeError("internal Stage-2 distilled adapter was not CPU-prepared before CUDA packing")
            pipe.load_lora_weights(
                str(stage2_local_path.parent),
                weight_name=stage2_local_path.name,
                adapter_name=build_policy.full_sft_stage2_adapter_name,
            )
            pipe.disable_lora()
            pipe.set_lora_device([build_policy.full_sft_stage2_adapter_name], device="cpu")
            stage2_record["effective"] = {
                "kind": "required_stage2_distilled_lora",
                "repo_id": build_policy.full_sft_stage2_lora_repo,
                "weight_name": weight_name,
                "revision": build_policy.full_sft_stage2_lora_revision,
                "resolved_revision": resolved_stage2_revision,
                "adapter_name": build_policy.full_sft_stage2_adapter_name,
                "strength": build_policy.full_sft_stage2_lora_strength,
                "residency": "startup CPU RAM-ready / Stage-2 GPU-lazy",
                "required_by_profile": True,
                "source_transport": "hf_hub_download/component-local",
                "size_bytes": int(stage2_local_path.stat().st_size),
            }
            log_fn(
                f"[D1R8P13] Full/SFT Stage-2 distilled LoRA startup RAM-ready "
                f"repo_id={build_policy.full_sft_stage2_lora_repo} revision={build_policy.full_sft_stage2_lora_revision!r} "
                f"strength={build_policy.full_sft_stage2_lora_strength}"
            )
        except Exception as exc:
            reason = f"{type(exc).__name__}: {exc}"
            stage2_record["fallback"] = True
            stage2_record["fallback_reason"] = reason
            stage2_record["effective"] = {"kind": "unavailable", "reason": reason}
            raise RuntimeError(f"Full/SFT Stage-2 distilled LoRA is required but failed to load: {reason}") from exc
        phases["full_sft_stage2_lora_ram_prepare_seconds"] = time.monotonic() - phase_started
    else:
        stage2_record["effective"] = {"kind": "not_applicable", "reason": "distilled_nf4 profile"}
    model_sources["stage2_distilled_lora"] = stage2_record

    # Optional heavy components remain CPU/RAM-ready at startup and GPU-lazy per request.
    diffusion_decode_pipe = None
    decoder_record = _source_record(None)
    decoder_record["requested"] = {
        "enabled": bool(build_policy.diffusion_decoder_enabled),
        "repo_id": repo_id,
        "path": "diffusion_decoder",
        "revision": revision,
    }
    if build_policy.diffusion_decoder_enabled:
        phase_started = time.monotonic()
        try:
            decoder = LTX2VideoDiffusionDecoderModel.from_pretrained(
                model_dir, subfolder="diffusion_decoder", dtype=torch.bfloat16
            )
            decoder.set_attn_processor(LTX2VideoVaeNeighborhoodNattenProcessor())
            decoder.enable_tiling()
            diffusion_decode_pipe = LTX2VideoDiffusionDecodePipeline(
                diffusion_decoder=decoder, scheduler=pipe.scheduler
            )
            decoder_record["effective"] = {
                "kind": "base_component",
                "repo_id": repo_id,
                "path": "diffusion_decoder",
                "revision": revision,
                "dtype": "bfloat16",
                "residency": "startup RAM-ready / GPU-lazy",
                "attention": "NATTEN via kernels",
                "tiling": True,
            }
            log_fn("[D1R8] diffusion decoder startup RAM-ready")
        except Exception as exc:
            decoder_record["fallback"] = True
            decoder_record["fallback_reason"] = f"{type(exc).__name__}: {exc}"
            decoder_record["effective"] = {"kind": "disabled", "reason": decoder_record["fallback_reason"]}
            log_fn(f"[D1R8] diffusion decoder disabled: {decoder_record['fallback_reason']}")
        phases["diffusion_decoder_ram_prepare_seconds"] = time.monotonic() - phase_started
    else:
        decoder_record["effective"] = {"kind": "disabled", "reason": "disabled by space_config.py"}
    model_sources["diffusion_decoder"] = decoder_record

    prompt_enhancer_model = None
    prompt_enhancer_processor = None
    phase_started = time.monotonic()
    try:
        prompt_enhancer_model, prompt_enhancer_processor = model_loading.load_prompt_enhancer_cpu(
            model_sources["prompt_enhancer"],
            repo_id=build_policy.prompt_enhancer_repo_id,
            revision=build_policy.prompt_enhancer_revision,
            enabled=build_policy.prompt_enhancer_enabled,
            policy=build_policy.prompt_enhancer_policy,
            token=token,
            log_fn=log_fn,
        )
    except Exception as exc:
        reason = f"{type(exc).__name__}: {exc}"
        record = model_sources["prompt_enhancer"]
        record["fallback"] = True
        record["fallback_reason"] = reason
        record["effective"] = {"kind": "disabled", "reason": reason}
        log_fn(f"[D1R8P3] prompt enhancer disabled: {reason}")
    phases["prompt_enhancer_ram_prepare_seconds"] = time.monotonic() - phase_started

    # Derive task-specific pipelines from the shared, already-packed base components.
    upsample_pipe = LTX2LatentUpsamplePipeline(vae=pipe.vae, latent_upsampler=latent_upsampler)
    pipe_i2v = LTX2ImageToVideoPipeline(
        scheduler=pipe.scheduler,
        vae=pipe.vae,
        audio_vae=pipe.audio_vae,
        text_encoder=pipe.text_encoder,
        tokenizer=pipe.tokenizer,
        connectors=pipe.connectors,
        transformer=pipe.transformer,
        vocoder=pipe.vocoder,
        processor=getattr(pipe, "processor", None),
        prompt_enhancer=None,
        duration_head=getattr(pipe, "duration_head", None),
    )
    pipe_condition = LTX2ConditionPipeline(
        scheduler=pipe.scheduler,
        audio_scheduler=getattr(pipe, "audio_scheduler", None),
        vae=pipe.vae,
        audio_vae=pipe.audio_vae,
        text_encoder=pipe.text_encoder,
        tokenizer=pipe.tokenizer,
        connectors=pipe.connectors,
        transformer=pipe.transformer,
        vocoder=pipe.vocoder,
        processor=getattr(pipe, "processor", None),
        prompt_enhancer=None,
        duration_head=getattr(pipe, "duration_head", None),
    )

    pipe_ic = None
    if build_policy.ic_colorizer_enabled or build_policy.ic_pixel_upscaler_enabled or build_policy.ic_inoutpaint_enabled:
        ic_scheduler = FlowMatchEulerDiscreteScheduler.from_config(
            pipe.scheduler.config, use_dynamic_shifting=False, shift_terminal=None
        )
        base_audio_scheduler = getattr(pipe, "audio_scheduler", None)
        ic_audio_scheduler = (
            FlowMatchEulerDiscreteScheduler.from_config(
                base_audio_scheduler.config, use_dynamic_shifting=False, shift_terminal=None
            )
            if base_audio_scheduler is not None
            else None
        )
        pipe_ic = LTX2InContextPipeline(
            scheduler=ic_scheduler,
            audio_scheduler=ic_audio_scheduler,
            vae=pipe.vae,
            audio_vae=pipe.audio_vae,
            text_encoder=pipe.text_encoder,
            tokenizer=pipe.tokenizer,
            connectors=pipe.connectors,
            transformer=pipe.transformer,
            vocoder=pipe.vocoder,
            processor=getattr(pipe, "processor", None),
            prompt_enhancer=None,
        )
    phases["attempt_total_seconds"] = time.monotonic() - attempt_started
    return RuntimeArtifacts(
        pipe=pipe,
        pipe_i2v=pipe_i2v,
        pipe_condition=pipe_condition,
        pipe_ic=pipe_ic,
        upsample_pipe=upsample_pipe,
        diffusion_decode_pipe=diffusion_decode_pipe,
        prompt_enhancer_model=prompt_enhancer_model,
        prompt_enhancer_processor=prompt_enhancer_processor,
        model_sources=model_sources,
        attention_state=attention_state,
        phases=phases,
        model_dir=model_dir,
    )


def try_runtime_for_base(*args, **kwargs):
    try:
        return attempt_runtime_for_base(*args, **kwargs), None
    except Exception as exc:
        return None, f"{type(exc).__name__}: {exc}"


def _runtime_contract(
    model_sources: dict,
    preload_policy: RuntimePreloadPolicy,
    pipe_ic: object | None,
) -> dict:
    return {
        "base_quantization_policy": (model_sources.get("base", {}).get("effective", {}) or {}).get(
            "quantization_policy"
        ),
        "runtime_profile": preload_policy.runtime_profile,
        "profile_switch": "space_config.py + restart only; no in-session model toggle",
        "transformer": model_sources["transformer"]["effective"],
        "stage2_distilled_lora": (model_sources.get("stage2_distilled_lora") or {}).get("effective"),
        "text_encoder": model_sources["text_encoder"]["effective"],
        "vae_audio_vocoder_connectors_upsampler": "BF16",
        "decoder": "Conv VAE tiled default; optional diffusion decoder startup RAM-ready / GPU-lazy",
        "lora": (
            "Full/SFT interop path: request-scoped user LoRAs active in Stage 1 and retained in Stage 2 alongside the internal distilled adapter; internal adapter preserved across user cleanup"
            if preload_policy.is_full_sft_profile
            else "built-in/session live adapters; never fused; selective request cleanup"
        ),
        "worker_isolation": "/tmp/ltx25_workers/<worker_uuid>/<session_uuid>/<request_uuid>/",
        "ssr_mode": False,
        "compile": False,
        "duration_head": (model_sources.get("duration_head") or {}).get("effective"),
        "prompt_enhancer": (model_sources.get("prompt_enhancer") or {}).get("effective"),
        "conditioning_modes": (
            "T2V / I2V / experimental FLF2V / experimental Start+Middle(+End) timeline keyframes "
            "via shared LTX2ConditionPipeline components"
        ),
        "ic_colorizer": (
            "dedicated LTX2InContextPipeline tab; shared LTX-2.5 components; independent schedulers; "
            "official LTX-2.3 Colorization adapter CPU-prepared then request-scoped on GPU; "
            + ("Full/SFT path exposed but not yet live-validated" if preload_policy.is_full_sft_profile else "Distilled live-closed stage-1-only product path")
            if pipe_ic is not None
            else "disabled by space_config.py"
        ),
        "ic_pixel_upscaler": (
            "official LTX-2.5 x2 Pixel Spatial Upscaler IC-LoRA; Distilled-only; reference prepared at target/2 and one request-scoped LTX2InContextPipeline pass renders the x2 target; binary adapter fetched from Hub rather than committed"
            if pipe_ic is not None and not preload_policy.is_full_sft_profile
            else "unavailable on this runtime profile or disabled by space_config.py"
        ),
        "ic_inoutpaint": (
            "official LTX-2.5 workflow compatibility gate reusing LTX-2.3 In-Outpainting IC-LoRA; Distilled-only; green-mask reference preprocess; request-scoped adapter; one-stage gate before official two-stage Laplacian parity"
            if pipe_ic is not None and not preload_policy.is_full_sft_profile
            else "unavailable on this runtime profile or disabled by space_config.py"
        ),
        "video_duration": (
            f"manual 1-15s standard; 15-30s opt-in experimental at 512x512 only; "
            f"Auto Duration experimental clamp {preload_policy.auto_duration_min_seconds:.1f}-"
            f"{preload_policy.auto_duration_max_seconds:.1f}s"
        ),
        "zerogpu_quota_ui": "merged into the Duration / ZeroGPU quota accordion when ZeroGPU runtime markers are detected",
        "zerogpu_duration_estimator": (
            "Full/SFT 90s validated 25f floor; >25f uses P27/P28 long-range base + P33 Diffusion-vs-Conv tail calibration, 110s minimum and 10%+5s guard"
            if preload_policy.is_full_sft_profile
            else "measured-runtime dynamic callable; diffusion decoder calibrated from 25f + 361f live timing"
        ),
    }


def preload_runtime(
    *,
    token: str | None,
    preload_policy: RuntimePreloadPolicy,
    build_policy: RuntimeBuildPolicy,
    log_fn: LogFn,
    disk_state_fn: Callable[[], dict],
    package_identity_fn: Callable[[str], dict],
    environment_identity_fn: Callable[[], dict],
) -> RuntimePreloadResult:
    """Build startup runtime state without owning application module globals."""
    started = time.monotonic()
    # Preload owns startup policy/fallback; app.py only assigns returned artifacts to globals.
    requested_base = {
        "repo_id": str(preload_policy.model_repo_id or "").strip(),
        "revision": str(preload_policy.model_revision or "").strip() or None,
    }
    base_record = _source_record(dict(requested_base))
    state = {
        "status": "starting",
        "candidate_id": preload_policy.candidate_id,
        "model_id": requested_base["repo_id"],
        "model_revision": requested_base["revision"],
        "quantization_policy_requested": preload_policy.model_quantization_policy_requested,
        "quantization_policy_effective": preload_policy.quantization_policy,
        "runtime_profile_requested": str(preload_policy.model_runtime_profile_requested),
        "runtime_profile_effective": preload_policy.runtime_profile,
        "config_warnings": list(preload_policy.config_warnings),
        "worker_uuid": preload_policy.worker_uuid,
        "disk_before": disk_state_fn(),
        "environment": environment_identity_fn(),
        "packages": {
            name: package_identity_fn(name)
            for name in (
                "torch",
                "diffusers",
                "transformers",
                "torchvision",
                "bitsandbytes",
                "accelerate",
                "peft",
                "gradio",
                "kernels",
                "gguf",
                "numpy",
                "scipy",
            )
        },
    }
    artifacts = None
    artifacts_assignable = False
    try:
        if not token:
            raise RuntimeError(
                "HF_TOKEN/HF_ACCESS_TOKEN is required because the canonical LTX-2.5 Diffusers repository is gated. "
                "Accept the upstream license and add a read token as a Space secret."
            )
        if not requested_base["repo_id"]:
            raise RuntimeError("MODEL_REPO_ID must not be empty.")

        for warning in preload_policy.config_warnings:
            log_fn(f"[SPACE_CONFIG] warning: {warning}")

        artifacts, configured_error = try_runtime_for_base(
            requested_base["repo_id"],
            requested_base["revision"],
            base_record,
            preload_policy.quantization_policy,
            token=token,
            build_policy=build_policy,
            log_fn=log_fn,
        )
        configured_is_safe_canonical = (
            requested_base["repo_id"] == preload_policy.canonical_model_repo_id
            and requested_base["revision"] == preload_policy.canonical_model_revision
            and preload_policy.quantization_policy == "nf4_auto"
        )
        if artifacts is None:
            if preload_policy.is_full_sft_profile:
                raise RuntimeError(
                    f"full_sft_nf4 exclusive profile failed and will not fall back to distilled: {configured_error}"
                )
            if configured_is_safe_canonical:
                raise RuntimeError(f"canonical NF4 runtime failed: {configured_error}")
            reason = str(configured_error)
            log_fn(f"[MODEL_OVERRIDE] configured base FAILED: {reason}")
            base_record = _source_record(dict(requested_base))
            base_record["fallback"] = True
            base_record["fallback_reason"] = reason
            log_fn(
                f"[MODEL_OVERRIDE] falling back to canonical NF4 base "
                f"{preload_policy.canonical_model_repo_id}@{preload_policy.canonical_model_revision}"
            )
            gc.collect()
            if torch.cuda.is_available():
                torch.cuda.empty_cache()
            artifacts, canonical_error = try_runtime_for_base(
                preload_policy.canonical_model_repo_id,
                preload_policy.canonical_model_revision,
                base_record,
                "nf4_auto",
                token=token,
                build_policy=build_policy,
                log_fn=log_fn,
            )
            if artifacts is None:
                raise RuntimeError(f"canonical NF4 fallback failed: {canonical_error}")

        pipe = artifacts.pipe
        pipe_ic = artifacts.pipe_ic
        model_sources = artifacts.model_sources
        attention_state = artifacts.attention_state
        phases = artifacts.phases
        model_dir = artifacts.model_dir
        artifacts_assignable = True
        state["disk_after_assets"] = disk_state_fn()
        state.update(
            status="ready",
            elapsed_seconds=time.monotonic() - started,
            quantization_policy_effective=(model_sources.get("base", {}).get("effective", {}) or {}).get(
                "quantization_policy", preload_policy.quantization_policy
            ),
            preload_phases=phases,
            model_dir=str(model_dir),
            model_sources=model_sources,
            attention_backend=attention_state,
            runtime_contract=_runtime_contract(model_sources, preload_policy, pipe_ic),
            runtime_mode=("zerogpu" if preload_policy.is_zerogpu else "non_zerogpu"),
        )
    except Exception as exc:
        if not artifacts_assignable:
            artifacts = None
        state.update(
            status="failed",
            elapsed_seconds=time.monotonic() - started,
            error_type=type(exc).__name__,
            error=str(exc),
        )
        log_fn(f"[D1] preload failed: {type(exc).__name__}: {exc}")
    log_fn(f"[D1] preload state: {json.dumps(state, sort_keys=True)}")
    return RuntimePreloadResult(runtime=artifacts, state=state)