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////", "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)