Runtime Loader API

RuntimeLoader owns materialization, runtime planning, and safe fallback during backend loading.

Loaded runtime bundle containing the finalized backend and plan metadata.

Attributes:

Name	Type	Description
resolved_model	ResolvedModel	Final resolved model metadata for the loaded runtime.
config	RuntimeConfig	Effective runtime configuration after selector application.
backend	BackendRuntime	Loaded backend runtime implementation.
model_path	Path | None	Local materialized model path when one exists.
plan	RuntimePlan	Final runtime plan used to load the backend.

Source code in src/ollm/runtime/loader.py

@dataclass(slots=True)
class LoadedRuntime:
    """Loaded runtime bundle containing the finalized backend and plan metadata.

    Attributes:
        resolved_model (ResolvedModel): Final resolved model metadata for the
            loaded runtime.
        config (RuntimeConfig): Effective runtime configuration after selector
            application.
        backend (BackendRuntime): Loaded backend runtime implementation.
        model_path (Path | None): Local materialized model path when one exists.
        plan (RuntimePlan): Final runtime plan used to load the backend.
    """

    resolved_model: ResolvedModel
    config: RuntimeConfig
    backend: BackendRuntime
    model_path: Path | None
    plan: RuntimePlan
    _kv_cache_instances: dict[tuple[Path, str, str, int | None], object] = field(
        default_factory=dict
    )

    @property
    def capabilities(self) -> CapabilityProfile:
        """Return capability information aligned with the finalized runtime plan.

        Returns:
            CapabilityProfile: Capability metadata adjusted to reflect the final
            support level and disk-cache behavior of the loaded runtime.
        """
        resolved_capabilities = self.resolved_model.capabilities
        if (
            resolved_capabilities.support_level is self.plan.support_level
            and resolved_capabilities.supports_disk_cache
            is self.plan.supports_disk_cache
        ):
            return resolved_capabilities
        details = dict(resolved_capabilities.details)
        details["runtime_plan_support_level"] = self.plan.support_level.value
        return CapabilityProfile(
            support_level=self.plan.support_level,
            modalities=resolved_capabilities.modalities,
            requires_processor=resolved_capabilities.requires_processor,
            supports_disk_cache=self.plan.supports_disk_cache,
            supports_local_materialization=resolved_capabilities.supports_local_materialization,
            supports_specialization=resolved_capabilities.supports_specialization,
            details=details,
        )

    @property
    def model(self):
        """Expose the backend-owned model object when one exists."""
        return self.backend.model

    @property
    def tokenizer(self):
        """Expose the backend-owned tokenizer when one exists."""
        return self.backend.tokenizer

    @property
    def processor(self):
        """Expose the backend-owned processor when one exists."""
        return self.backend.processor

    @property
    def device(self) -> torch.device:
        """Expose the backend runtime device."""
        return self.backend.device

    def get_or_create_kv_cache(
        self,
        cache_dir: Path,
        strategy: str,
        lifecycle: str,
        window_tokens: int | None,
    ) -> object | None:
        """Reuse one KV-cache instance per resolved cache key.

        Args:
            cache_dir (Path): Cache root for the KV cache instance.
            strategy (str): Resolved KV cache strategy ID.
            lifecycle (str): Resolved cache lifecycle ID.
            window_tokens (int | None): Sliding-window token budget when the
                strategy requires one.

        Returns:
            object | None: Existing or newly created cache object, or ``None``
            when the backend does not expose a cache.
        """
        cache_key = (cache_dir.resolve(), strategy, lifecycle, window_tokens)
        cache = self._kv_cache_instances.get(cache_key)
        if cache is not None:
            return cache
        created_cache = self.backend.create_cache(
            cache_dir, strategy, lifecycle, window_tokens
        )
        if created_cache is not None:
            self._kv_cache_instances[cache_key] = created_cache
        return created_cache

    def reset_kv_cache_instances(self) -> None:
        """Drop any cached KV objects before a full-history re-execution."""

        self._kv_cache_instances.clear()

capabilities property

capabilities: CapabilityProfile

Return capability information aligned with the finalized runtime plan.

Returns:

Name	Type	Description
CapabilityProfile	CapabilityProfile	Capability metadata adjusted to reflect the final
	CapabilityProfile	support level and disk-cache behavior of the loaded runtime.

model property

model

Expose the backend-owned model object when one exists.

tokenizer property

tokenizer

Expose the backend-owned tokenizer when one exists.

processor property

processor

Expose the backend-owned processor when one exists.

device property

device: device

Expose the backend runtime device.

get_or_create_kv_cache

get_or_create_kv_cache(
    cache_dir: Path,
    strategy: str,
    lifecycle: str,
    window_tokens: int | None,
) -> object | None

Reuse one KV-cache instance per resolved cache key.

Parameters:

Name	Type	Description	Default
cache_dir	Path	Cache root for the KV cache instance.	required
strategy	str	Resolved KV cache strategy ID.	required
lifecycle	str	Resolved cache lifecycle ID.	required
window_tokens	int | None	Sliding-window token budget when the strategy requires one.	required

Returns:

Type	Description
object | None	object | None: Existing or newly created cache object, or None
object | None	when the backend does not expose a cache.

Source code in src/ollm/runtime/loader.py

def get_or_create_kv_cache(
    self,
    cache_dir: Path,
    strategy: str,
    lifecycle: str,
    window_tokens: int | None,
) -> object | None:
    """Reuse one KV-cache instance per resolved cache key.

    Args:
        cache_dir (Path): Cache root for the KV cache instance.
        strategy (str): Resolved KV cache strategy ID.
        lifecycle (str): Resolved cache lifecycle ID.
        window_tokens (int | None): Sliding-window token budget when the
            strategy requires one.

    Returns:
        object | None: Existing or newly created cache object, or ``None``
        when the backend does not expose a cache.
    """
    cache_key = (cache_dir.resolve(), strategy, lifecycle, window_tokens)
    cache = self._kv_cache_instances.get(cache_key)
    if cache is not None:
        return cache
    created_cache = self.backend.create_cache(
        cache_dir, strategy, lifecycle, window_tokens
    )
    if created_cache is not None:
        self._kv_cache_instances[cache_key] = created_cache
    return created_cache

reset_kv_cache_instances

reset_kv_cache_instances() -> None

Drop any cached KV objects before a full-history re-execution.

Source code in src/ollm/runtime/loader.py

def reset_kv_cache_instances(self) -> None:
    """Drop any cached KV objects before a full-history re-execution."""

    self._kv_cache_instances.clear()

Resolve, plan, materialize, and load runtimes for model references.

Parameters:

Name	Type	Description	Default
resolver	ModelResolver | None	Optional resolver override.	None
selector	BackendSelector | None	Optional backend selector override.	None
backends	tuple[ExecutionBackend, ...] | None	Optional registered backend implementations.	None
snapshot_downloader	Callable[[str, str, bool, str | None], None] | None	Optional Hugging Face snapshot downloader override.	None
specialization_registry	SpecializationRegistry | None	Optional specialization registry override.	None

Source code in src/ollm/runtime/loader.py

class RuntimeLoader:
    """Resolve, plan, materialize, and load runtimes for model references.

    Args:
        resolver (ModelResolver | None): Optional resolver override.
        selector (BackendSelector | None): Optional backend selector override.
        backends (tuple[ExecutionBackend, ...] | None): Optional registered
            backend implementations.
        snapshot_downloader (Callable[[str, str, bool, str | None], None] | None):
            Optional Hugging Face snapshot downloader override.
        specialization_registry (SpecializationRegistry | None): Optional
            specialization registry override.
    """

    def __init__(
        self,
        resolver: ModelResolver | None = None,
        selector: BackendSelector | None = None,
        backends: tuple[ExecutionBackend, ...] | None = None,
        snapshot_downloader: Callable[[str, str, bool, str | None], None] | None = None,
        specialization_registry: SpecializationRegistry | None = None,
    ):
        self._resolver = resolver or ModelResolver()
        self._specialization_registry = (
            build_default_specialization_registry()
            if specialization_registry is None
            else specialization_registry
        )
        self._selector = selector or BackendSelector(
            specialization_registry=self._specialization_registry
        )
        backend_list = backends or (
            NativeOptimizedBackend(
                specialization_registry=self._specialization_registry
            ),
            TransformersGenericBackend(),
        )
        self._backends = {backend.backend_id: backend for backend in backend_list}
        self._snapshot_downloader = (
            download_hf_snapshot if snapshot_downloader is None else snapshot_downloader
        )

    def resolve(self, model_reference: str, models_dir: Path) -> ResolvedModel:
        """Resolve a model reference without planning or loading.

        Args:
            model_reference (str): User-facing model reference.
            models_dir (Path): Local models root used for implicit path
                resolution.

        Returns:
            ResolvedModel: Normalized model metadata.
        """
        return self._resolver.resolve(model_reference, models_dir)

    def discover_local_models(self, models_dir: Path) -> tuple[ResolvedModel, ...]:
        """Discover local materialized models under a models directory.

        Args:
            models_dir (Path): Local models root to inspect.

        Returns:
            tuple[ResolvedModel, ...]: Resolved local model directories found
            under the given root.
        """
        return self._resolver.discover_local_models(models_dir)

    def download(
        self, model_reference: str, models_dir: Path, force_download: bool = False
    ) -> Path:
        """Materialize a downloadable model reference locally.

        Args:
            model_reference (str): User-facing model reference to materialize.
            models_dir (Path): Local models root used for materialization.
            force_download (bool): Whether to re-download even when a managed
                directory already exists.

        Returns:
            Path: Local materialized model directory.

        Raises:
            ValueError: Raised when the reference cannot be materialized or the
                resulting managed directory is incomplete.
        """
        resolved_model = self.resolve(model_reference, models_dir)
        if resolved_model.source_kind is ModelSourceKind.LOCAL_PATH:
            if (
                resolved_model.model_path is None
                or not resolved_model.model_path.exists()
            ):
                raise ValueError(f"Local model path does not exist: {model_reference}")
            return resolved_model.model_path
        if (
            not resolved_model.is_downloadable()
            or resolved_model.repo_id is None
            or resolved_model.model_path is None
        ):
            raise ValueError(
                f"Model reference '{model_reference}' does not support snapshot download"
            )
        if resolved_model.model_path.exists() and not force_download:
            self._repair_managed_model_dir(resolved_model)
            if self._managed_model_dir_is_complete(resolved_model):
                return resolved_model.model_path
        resolved_model.model_path.parent.mkdir(parents=True, exist_ok=True)
        self._snapshot_downloader(
            resolved_model.repo_id,
            str(resolved_model.model_path),
            force_download,
            resolved_model.revision,
        )
        self._repair_managed_model_dir(resolved_model)
        self._validate_managed_model_dir(resolved_model)
        return resolved_model.model_path

    def load(self, config: RuntimeConfig) -> LoadedRuntime:
        """Validate, plan, and load a runtime backend for execution.

        Args:
            config (RuntimeConfig): Runtime configuration to execute.

        Returns:
            LoadedRuntime: Loaded backend runtime bundle ready for execution.

        Raises:
            ValueError: Raised when planning, materialization, specialization, or
                backend loading fails without a truthful fallback path.
        """
        config.validate()
        resolved_model = self.resolve(
            config.model_reference, config.resolved_models_dir()
        )
        if (
            resolved_model.model_path is None
            and resolved_model.capabilities.support_level is SupportLevel.UNSUPPORTED
        ):
            raise ValueError(resolved_model.resolution_message)
        model_path = self._ensure_local_model(resolved_model, config.force_download)
        execution_model = self._refresh_materialized_model(resolved_model, model_path)

        preliminary_plan = self._refine_runtime_plan(
            self._selector.select(execution_model, config), config
        )
        strategy_selection = select_runtime_strategy(
            resolved_model=execution_model,
            runtime_plan=preliminary_plan,
            requested_strategy_override=config.requested_kv_cache_strategy(),
            strategy_selector_profile=config.resolved_strategy_selector_profile(),
            requested_window_tokens=config.kv_cache_window_tokens,
        )
        effective_config = apply_strategy_selection_to_config(
            config, strategy_selection
        )
        runtime_plan = plan_with_strategy_selection(
            preliminary_plan,
            strategy_selection,
            requested_config=config,
            effective_config=effective_config,
        )
        if not runtime_plan.is_executable():
            raise ValueError(runtime_plan.reason)
        self._validate_runtime_plan(runtime_plan, effective_config)
        try:
            backend_runtime = self._load_backend_runtime(runtime_plan, effective_config)
            runtime_plan = self._finalize_runtime_plan(runtime_plan, backend_runtime)
        except (SpecializationApplicationError, SpecializationLoadError) as exc:
            fallback_plan = self._build_generic_fallback_plan(
                execution_model=execution_model,
                runtime_plan=runtime_plan,
                config=effective_config,
                error=exc,
            )
            if fallback_plan is None:
                raise ValueError(str(exc)) from exc
            strategy_selection = select_runtime_strategy(
                resolved_model=execution_model,
                runtime_plan=fallback_plan,
                requested_strategy_override=config.requested_kv_cache_strategy(),
                strategy_selector_profile=config.resolved_strategy_selector_profile(),
                requested_window_tokens=config.kv_cache_window_tokens,
            )
            effective_config = apply_strategy_selection_to_config(
                config, strategy_selection
            )
            fallback_plan = plan_with_strategy_selection(
                fallback_plan,
                strategy_selection,
                requested_config=config,
                effective_config=effective_config,
            )
            backend_runtime = self._load_backend_runtime(
                fallback_plan, effective_config
            )
            runtime_plan = self._finalize_runtime_plan(fallback_plan, backend_runtime)
        return LoadedRuntime(
            resolved_model=runtime_plan.resolved_model,
            config=effective_config,
            backend=backend_runtime,
            model_path=runtime_plan.model_path,
            plan=runtime_plan,
        )

    def plan(self, config: RuntimeConfig) -> RuntimePlan:
        """Build a runtime plan without loading a backend.

        Args:
            config (RuntimeConfig): Runtime configuration to inspect.

        Returns:
            RuntimePlan: Planned backend, specialization, and strategy state.

        Raises:
            ValueError: Raised when the runtime configuration is invalid or no
                truthful plan can be produced.
        """
        config.validate()
        resolved_model = self.resolve(
            config.model_reference, config.resolved_models_dir()
        )
        if resolved_model.model_path is None or not resolved_model.model_path.exists():
            preliminary_plan = self._refine_runtime_plan(
                self._selector.select(resolved_model, config), config
            )
            strategy_selection = select_runtime_strategy(
                resolved_model=resolved_model,
                runtime_plan=preliminary_plan,
                requested_strategy_override=config.requested_kv_cache_strategy(),
                strategy_selector_profile=config.resolved_strategy_selector_profile(),
                requested_window_tokens=config.kv_cache_window_tokens,
            )
            effective_config = apply_strategy_selection_to_config(
                config, strategy_selection
            )
            return plan_with_strategy_selection(
                preliminary_plan,
                strategy_selection,
                requested_config=config,
                effective_config=effective_config,
            )
        execution_model = self._refresh_materialized_model(
            resolved_model, resolved_model.model_path
        )
        preliminary_plan = self._refine_runtime_plan(
            self._selector.select(execution_model, config), config
        )
        strategy_selection = select_runtime_strategy(
            resolved_model=execution_model,
            runtime_plan=preliminary_plan,
            requested_strategy_override=config.requested_kv_cache_strategy(),
            strategy_selector_profile=config.resolved_strategy_selector_profile(),
            requested_window_tokens=config.kv_cache_window_tokens,
        )
        effective_config = apply_strategy_selection_to_config(
            config, strategy_selection
        )
        return plan_with_strategy_selection(
            preliminary_plan,
            strategy_selection,
            requested_config=config,
            effective_config=effective_config,
        )

    def _refresh_materialized_model(
        self, resolved_model: ResolvedModel, model_path: Path
    ) -> ResolvedModel:
        return self._resolver.inspect_materialized_model(
            resolved_model.reference,
            model_path,
            source_kind=resolved_model.source_kind,
            repo_id=resolved_model.repo_id,
            revision=resolved_model.revision,
            catalog_entry=resolved_model.catalog_entry,
        )

    def _ensure_local_model(
        self, resolved_model: ResolvedModel, force_download: bool
    ) -> Path:
        if resolved_model.model_path is None:
            raise ValueError(
                f"Model reference '{resolved_model.reference.raw}' does not resolve to a local model path"
            )
        if resolved_model.model_path.exists() and not force_download:
            self._repair_managed_model_dir(resolved_model)
            if self._managed_model_dir_is_complete(resolved_model):
                return resolved_model.model_path
        if resolved_model.repo_id is None:
            raise ValueError(
                f"Local model path does not exist: {resolved_model.model_path}"
            )
        resolved_model.model_path.parent.mkdir(parents=True, exist_ok=True)
        self._snapshot_downloader(
            resolved_model.repo_id,
            str(resolved_model.model_path),
            force_download,
            resolved_model.revision,
        )
        self._repair_managed_model_dir(resolved_model)
        self._validate_managed_model_dir(resolved_model)
        return resolved_model.model_path

    def _repair_managed_model_dir(self, resolved_model: ResolvedModel) -> None:
        model_path = resolved_model.model_path
        if (
            resolved_model.repo_id is None
            or model_path is None
            or not model_path.exists()
        ):
            return
        prune_hf_runtime_artifacts(model_path)

    def _managed_model_dir_is_complete(self, resolved_model: ResolvedModel) -> bool:
        model_path = resolved_model.model_path
        if resolved_model.repo_id is None or model_path is None:
            return True
        return hf_runtime_artifacts_complete(model_path)

    def _validate_managed_model_dir(self, resolved_model: ResolvedModel) -> None:
        model_path = resolved_model.model_path
        if model_path is None:
            return
        artifact_gaps = _runtime_artifact_gaps(model_path)
        if not artifact_gaps:
            return
        raise ValueError(
            "Managed model directory is missing required runtime artifacts: "
            f"{model_path} ({'; '.join(artifact_gaps)})"
        )

    def _validate_runtime_plan(
        self, runtime_plan: RuntimePlan, config: RuntimeConfig
    ) -> None:
        if config.offload_gpu_layers > 0 and not runtime_plan.supports_gpu_offload:
            raise ValueError(
                f"Selected backend '{runtime_plan.backend_id}' does not support GPU layer offload controls"
            )
        if config.offload_cpu_layers > 0 and not runtime_plan.supports_cpu_offload:
            raise ValueError(
                f"Selected backend '{runtime_plan.backend_id}' does not support CPU layer offload controls"
            )

    def _load_backend_runtime(
        self, runtime_plan: RuntimePlan, config: RuntimeConfig
    ) -> BackendRuntime:
        backend_id = runtime_plan.backend_id
        if backend_id is None:
            raise ValueError("Runtime plan did not select a backend")
        backend_impl = self._backends.get(backend_id)
        if backend_impl is None:
            raise ValueError(f"No runtime backend is registered for '{backend_id}'")
        backend_runtime = backend_impl.load(runtime_plan, config)
        backend_runtime.apply_offload(config)
        return backend_runtime

    def _refine_runtime_plan(
        self, runtime_plan: RuntimePlan, config: RuntimeConfig
    ) -> RuntimePlan:
        if runtime_plan.backend_id is None:
            return runtime_plan
        backend_impl = self._backends.get(runtime_plan.backend_id)
        if backend_impl is None:
            return runtime_plan
        return backend_impl.refine_plan(runtime_plan, config)

    def _finalize_runtime_plan(
        self, runtime_plan: RuntimePlan, backend_runtime: BackendRuntime
    ) -> RuntimePlan:
        details = dict(runtime_plan.details)
        for key, value in backend_runtime.details.items():
            details[key] = value
        if backend_runtime.applied_specialization is None:
            return replace(runtime_plan, details=details)
        applied_specialization = backend_runtime.applied_specialization
        details["applied_specialization_pass_ids"] = ",".join(
            pass_id.value for pass_id in applied_specialization.applied_pass_ids
        )
        for key, value in applied_specialization.details.items():
            details[key] = value
        return replace(
            runtime_plan,
            specialization_applied=True,
            specialization_state=SpecializationState.APPLIED,
            applied_specialization_pass_ids=applied_specialization.applied_pass_ids,
            details=details,
        )

    def _build_generic_fallback_plan(
        self,
        execution_model: ResolvedModel,
        runtime_plan: RuntimePlan,
        config: RuntimeConfig,
        error: SpecializationApplicationError | SpecializationLoadError,
    ) -> RuntimePlan | None:
        if runtime_plan.backend_id != "optimized-native":
            return None
        if config.resolved_backend() is not None:
            return None
        if execution_model.generic_model_kind is None:
            return None
        if "transformers-generic" not in self._backends:
            return None
        if config.offload_cpu_layers > 0 or config.offload_gpu_layers > 0:
            return None

        details = dict(runtime_plan.details)
        details["fallback_from_backend_id"] = runtime_plan.backend_id or "unknown"
        details["fallback_error_type"] = type(error).__name__
        details["fallback_provider_id"] = (
            runtime_plan.specialization_provider_id
            or getattr(error, "provider_id", "unknown")
        )
        reason = (
            f"Fell back to transformers-generic for {execution_model.reference.raw} after optimized "
            f"specialization failed: {error}"
        )
        return RuntimePlan(
            resolved_model=execution_model,
            backend_id="transformers-generic",
            model_path=execution_model.model_path,
            support_level=SupportLevel.GENERIC,
            generic_model_kind=execution_model.generic_model_kind,
            supports_disk_cache=False,
            supports_cpu_offload=False,
            supports_gpu_offload=False,
            specialization_enabled=runtime_plan.specialization_enabled,
            specialization_applied=False,
            specialization_provider_id=runtime_plan.specialization_provider_id,
            specialization_state=SpecializationState.FALLBACK,
            reason=reason,
            specialization_pass_ids=runtime_plan.specialization_pass_ids,
            applied_specialization_pass_ids=(),
            fallback_reason=str(error),
            details=details,
        )

resolve

resolve(
    model_reference: str, models_dir: Path
) -> ResolvedModel

Resolve a model reference without planning or loading.

Parameters:

Name	Type	Description	Default
model_reference	str	User-facing model reference.	required
models_dir	Path	Local models root used for implicit path resolution.	required

Returns:

Name	Type	Description
ResolvedModel	ResolvedModel	Normalized model metadata.

Source code in src/ollm/runtime/loader.py

def resolve(self, model_reference: str, models_dir: Path) -> ResolvedModel:
    """Resolve a model reference without planning or loading.

    Args:
        model_reference (str): User-facing model reference.
        models_dir (Path): Local models root used for implicit path
            resolution.

    Returns:
        ResolvedModel: Normalized model metadata.
    """
    return self._resolver.resolve(model_reference, models_dir)

discover_local_models

discover_local_models(
    models_dir: Path,
) -> tuple[ResolvedModel, ...]

Discover local materialized models under a models directory.

Parameters:

Name	Type	Description	Default
models_dir	Path	Local models root to inspect.	required

Returns:

Type	Description
ResolvedModel	tuple[ResolvedModel, ...]: Resolved local model directories found
...	under the given root.

Source code in src/ollm/runtime/loader.py

def discover_local_models(self, models_dir: Path) -> tuple[ResolvedModel, ...]:
    """Discover local materialized models under a models directory.

    Args:
        models_dir (Path): Local models root to inspect.

    Returns:
        tuple[ResolvedModel, ...]: Resolved local model directories found
        under the given root.
    """
    return self._resolver.discover_local_models(models_dir)

download

download(
    model_reference: str,
    models_dir: Path,
    force_download: bool = False,
) -> Path

Materialize a downloadable model reference locally.

Parameters:

Name	Type	Description	Default
model_reference	str	User-facing model reference to materialize.	required
models_dir	Path	Local models root used for materialization.	required
force_download	bool	Whether to re-download even when a managed directory already exists.	False

Returns:

Name	Type	Description
Path	Path	Local materialized model directory.

Raises:

Type	Description
ValueError	Raised when the reference cannot be materialized or the resulting managed directory is incomplete.

Source code in src/ollm/runtime/loader.py

def download(
    self, model_reference: str, models_dir: Path, force_download: bool = False
) -> Path:
    """Materialize a downloadable model reference locally.

    Args:
        model_reference (str): User-facing model reference to materialize.
        models_dir (Path): Local models root used for materialization.
        force_download (bool): Whether to re-download even when a managed
            directory already exists.

    Returns:
        Path: Local materialized model directory.

    Raises:
        ValueError: Raised when the reference cannot be materialized or the
            resulting managed directory is incomplete.
    """
    resolved_model = self.resolve(model_reference, models_dir)
    if resolved_model.source_kind is ModelSourceKind.LOCAL_PATH:
        if (
            resolved_model.model_path is None
            or not resolved_model.model_path.exists()
        ):
            raise ValueError(f"Local model path does not exist: {model_reference}")
        return resolved_model.model_path
    if (
        not resolved_model.is_downloadable()
        or resolved_model.repo_id is None
        or resolved_model.model_path is None
    ):
        raise ValueError(
            f"Model reference '{model_reference}' does not support snapshot download"
        )
    if resolved_model.model_path.exists() and not force_download:
        self._repair_managed_model_dir(resolved_model)
        if self._managed_model_dir_is_complete(resolved_model):
            return resolved_model.model_path
    resolved_model.model_path.parent.mkdir(parents=True, exist_ok=True)
    self._snapshot_downloader(
        resolved_model.repo_id,
        str(resolved_model.model_path),
        force_download,
        resolved_model.revision,
    )
    self._repair_managed_model_dir(resolved_model)
    self._validate_managed_model_dir(resolved_model)
    return resolved_model.model_path

load

load(config: RuntimeConfig) -> LoadedRuntime

Validate, plan, and load a runtime backend for execution.

Parameters:

Name	Type	Description	Default
config	RuntimeConfig	Runtime configuration to execute.	required

Returns:

Name	Type	Description
LoadedRuntime	LoadedRuntime	Loaded backend runtime bundle ready for execution.

Raises:

Type	Description
ValueError	Raised when planning, materialization, specialization, or backend loading fails without a truthful fallback path.

Source code in src/ollm/runtime/loader.py

def load(self, config: RuntimeConfig) -> LoadedRuntime:
    """Validate, plan, and load a runtime backend for execution.

    Args:
        config (RuntimeConfig): Runtime configuration to execute.

    Returns:
        LoadedRuntime: Loaded backend runtime bundle ready for execution.

    Raises:
        ValueError: Raised when planning, materialization, specialization, or
            backend loading fails without a truthful fallback path.
    """
    config.validate()
    resolved_model = self.resolve(
        config.model_reference, config.resolved_models_dir()
    )
    if (
        resolved_model.model_path is None
        and resolved_model.capabilities.support_level is SupportLevel.UNSUPPORTED
    ):
        raise ValueError(resolved_model.resolution_message)
    model_path = self._ensure_local_model(resolved_model, config.force_download)
    execution_model = self._refresh_materialized_model(resolved_model, model_path)

    preliminary_plan = self._refine_runtime_plan(
        self._selector.select(execution_model, config), config
    )
    strategy_selection = select_runtime_strategy(
        resolved_model=execution_model,
        runtime_plan=preliminary_plan,
        requested_strategy_override=config.requested_kv_cache_strategy(),
        strategy_selector_profile=config.resolved_strategy_selector_profile(),
        requested_window_tokens=config.kv_cache_window_tokens,
    )
    effective_config = apply_strategy_selection_to_config(
        config, strategy_selection
    )
    runtime_plan = plan_with_strategy_selection(
        preliminary_plan,
        strategy_selection,
        requested_config=config,
        effective_config=effective_config,
    )
    if not runtime_plan.is_executable():
        raise ValueError(runtime_plan.reason)
    self._validate_runtime_plan(runtime_plan, effective_config)
    try:
        backend_runtime = self._load_backend_runtime(runtime_plan, effective_config)
        runtime_plan = self._finalize_runtime_plan(runtime_plan, backend_runtime)
    except (SpecializationApplicationError, SpecializationLoadError) as exc:
        fallback_plan = self._build_generic_fallback_plan(
            execution_model=execution_model,
            runtime_plan=runtime_plan,
            config=effective_config,
            error=exc,
        )
        if fallback_plan is None:
            raise ValueError(str(exc)) from exc
        strategy_selection = select_runtime_strategy(
            resolved_model=execution_model,
            runtime_plan=fallback_plan,
            requested_strategy_override=config.requested_kv_cache_strategy(),
            strategy_selector_profile=config.resolved_strategy_selector_profile(),
            requested_window_tokens=config.kv_cache_window_tokens,
        )
        effective_config = apply_strategy_selection_to_config(
            config, strategy_selection
        )
        fallback_plan = plan_with_strategy_selection(
            fallback_plan,
            strategy_selection,
            requested_config=config,
            effective_config=effective_config,
        )
        backend_runtime = self._load_backend_runtime(
            fallback_plan, effective_config
        )
        runtime_plan = self._finalize_runtime_plan(fallback_plan, backend_runtime)
    return LoadedRuntime(
        resolved_model=runtime_plan.resolved_model,
        config=effective_config,
        backend=backend_runtime,
        model_path=runtime_plan.model_path,
        plan=runtime_plan,
    )

plan

plan(config: RuntimeConfig) -> RuntimePlan

Build a runtime plan without loading a backend.

Parameters:

Name	Type	Description	Default
config	RuntimeConfig	Runtime configuration to inspect.	required

Returns:

Name	Type	Description
RuntimePlan	RuntimePlan	Planned backend, specialization, and strategy state.

Raises:

Type	Description
ValueError	Raised when the runtime configuration is invalid or no truthful plan can be produced.

Source code in src/ollm/runtime/loader.py

def plan(self, config: RuntimeConfig) -> RuntimePlan:
    """Build a runtime plan without loading a backend.

    Args:
        config (RuntimeConfig): Runtime configuration to inspect.

    Returns:
        RuntimePlan: Planned backend, specialization, and strategy state.

    Raises:
        ValueError: Raised when the runtime configuration is invalid or no
            truthful plan can be produced.
    """
    config.validate()
    resolved_model = self.resolve(
        config.model_reference, config.resolved_models_dir()
    )
    if resolved_model.model_path is None or not resolved_model.model_path.exists():
        preliminary_plan = self._refine_runtime_plan(
            self._selector.select(resolved_model, config), config
        )
        strategy_selection = select_runtime_strategy(
            resolved_model=resolved_model,
            runtime_plan=preliminary_plan,
            requested_strategy_override=config.requested_kv_cache_strategy(),
            strategy_selector_profile=config.resolved_strategy_selector_profile(),
            requested_window_tokens=config.kv_cache_window_tokens,
        )
        effective_config = apply_strategy_selection_to_config(
            config, strategy_selection
        )
        return plan_with_strategy_selection(
            preliminary_plan,
            strategy_selection,
            requested_config=config,
            effective_config=effective_config,
        )
    execution_model = self._refresh_materialized_model(
        resolved_model, resolved_model.model_path
    )
    preliminary_plan = self._refine_runtime_plan(
        self._selector.select(execution_model, config), config
    )
    strategy_selection = select_runtime_strategy(
        resolved_model=execution_model,
        runtime_plan=preliminary_plan,
        requested_strategy_override=config.requested_kv_cache_strategy(),
        strategy_selector_profile=config.resolved_strategy_selector_profile(),
        requested_window_tokens=config.kv_cache_window_tokens,
    )
    effective_config = apply_strategy_selection_to_config(
        config, strategy_selection
    )
    return plan_with_strategy_selection(
        preliminary_plan,
        strategy_selection,
        requested_config=config,
        effective_config=effective_config,
    )