jundot--omlx/omlx/settings.py

# SPDX-License-Identifier: Apache-2.0
"""
Global settings management for oMLX.

This module provides a centralized settings system with:
- Hierarchical configuration (CLI > env > file > defaults)
- Automatic directory creation
- System resource detection (RAM, SSD capacity)
- Settings persistence to JSON file

Usage:
    from omlx.settings import init_settings, get_settings

    # At startup
    init_settings(cli_args=args)

    # Anywhere else
    settings = get_settings()
    print(settings.server.port)
"""

from __future__ import annotations

import json
import logging
import os
import shutil
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal

from .config import parse_size

if TYPE_CHECKING:
    from .scheduler import SchedulerConfig

logger = logging.getLogger(__name__)

# Settings file version for future migrations
SETTINGS_VERSION = "1.0"

# Default base path
DEFAULT_BASE_PATH = Path.home() / ".omlx"

# One-line bootstrap file the macOS app writes when the user moves their data root
BASE_PATH_BOOTSTRAP_FILE = (
    Path.home() / "Library" / "Application Support" / "oMLX" / "base-path"
)


def resolve_default_base_path() -> Path:
    """
    Resolve the base path to use when none was passed explicitly.

    Priority: ``OMLX_BASE_PATH`` env var > the macOS app's bootstrap file >
    ``~/.omlx``. This matches AppConfig.currentBasePath() in the Swift app
    so the CLI and GUI agree on where settings.json lives.
    """
    env_value = os.environ.get("OMLX_BASE_PATH")
    if env_value:
        return Path(env_value).expanduser().resolve()

    try:
        raw = BASE_PATH_BOOTSTRAP_FILE.read_text(encoding="utf-8").strip()
    except OSError:
        raw = ""
    if raw:
        return Path(raw).expanduser().resolve()

    return DEFAULT_BASE_PATH


def get_system_memory() -> int:
    """
    Return total system RAM in bytes.

    Uses os.sysconf first, then psutil_compat so macOS does not depend on
    psutil's VM stats adapter, which can lag new HOST_VM_INFO64 layouts.

    Returns:
        Total RAM in bytes.
    """
    try:
        pages = os.sysconf("SC_PHYS_PAGES")
        page_size = os.sysconf("SC_PAGE_SIZE")
        memory = int(pages) * int(page_size)
        if memory > 0:
            return memory
    except (AttributeError, ValueError, OSError):
        pass

    try:
        from .utils import psutil_compat

        memory = int(psutil_compat.get_total_memory())
        if memory > 0:
            return memory
    except Exception as exc:  # noqa: BLE001
        logger.warning("psutil_compat failed to detect system memory: %s", exc)

    # Default to 16GB if detection fails
    logger.warning("Could not detect system memory, defaulting to 16GB")
    return 16 * 1024**3


def get_ssd_capacity(path: str | Path) -> int:
    """
    Return disk capacity in bytes for the given path.

    Args:
        path: Path to check disk capacity for.

    Returns:
        Total disk capacity in bytes.
    """
    path = Path(path).expanduser().resolve()

    # Ensure parent directory exists for capacity check
    check_path = path
    while not check_path.exists() and check_path.parent != check_path:
        check_path = check_path.parent

    try:
        usage = shutil.disk_usage(check_path)
        return usage.total
    except OSError as e:
        logger.warning(f"Could not get disk capacity for {path}: {e}")
        # Default to 500GB if detection fails
        return 500 * 1024**3


# Burst Decode UI modes -> (decode_burst_max_steps, decode_burst_budget_single_s).
# These mirror the OMLX_DECODE_BURST_* env vars read by EngineConfig
# (engine_core.py). "off" fully disables bursting via max_steps=1; the on-levels
# keep the default step cap and set the single-request time budget that controls
# how many decode steps coalesce per event-loop hand-off (higher = faster, but
# tokens stream in larger chunks).
BURST_DECODE_MODES: dict[str, tuple[int, float]] = {
    "off": (1, 0.0),
    "light": (64, 0.05),
    "balanced": (64, 0.1),
    "aggressive": (64, 0.2),
}
DEFAULT_BURST_DECODE_MODE = "balanced"


def burst_decode_env(mode: str) -> dict[str, str]:
    """Map a Burst Decode mode to the OMLX_DECODE_BURST_* env vars.

    EngineConfig reads these at construction, so seeding them lets engines
    loaded later pick up the mode without a server restart. An unknown mode
    falls back to the default so a stale settings.json never disables bursting
    unexpectedly.
    """
    max_steps, single_s = BURST_DECODE_MODES.get(
        mode, BURST_DECODE_MODES[DEFAULT_BURST_DECODE_MODE]
    )
    return {
        "OMLX_DECODE_BURST_MAX_STEPS": str(max_steps),
        "OMLX_DECODE_BURST_BUDGET_SINGLE_S": str(single_s),
    }


@dataclass
class ServerSettings:
    """Server configuration settings."""

    host: str = "127.0.0.1"
    port: int = 8000
    log_level: str = "info"
    cors_origins: list[str] = field(default_factory=lambda: ["*"])
    server_aliases: list[str] = field(default_factory=list)
    sse_keepalive_mode: str = "chunk"
    auto_start_on_launch: bool = True
    burst_decode_mode: str = DEFAULT_BURST_DECODE_MODE
    preserve_mid_system_cache: bool = True

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return asdict(self)

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> ServerSettings:
        """Create from dictionary."""
        _host = data.get("host", data.get("bind_address", "127.0.0.1"))
        return cls(
            host=", ".join(_host) if isinstance(_host, list) else str(_host),
            port=data.get("port", 8000),
            log_level=data.get("log_level", "info"),
            cors_origins=data.get("cors_origins", ["*"]),
            server_aliases=data.get("server_aliases", []),
            sse_keepalive_mode=data.get("sse_keepalive_mode", "chunk"),
            auto_start_on_launch=data.get("auto_start_on_launch", True),
            burst_decode_mode=data.get("burst_decode_mode", DEFAULT_BURST_DECODE_MODE),
            preserve_mid_system_cache=data.get("preserve_mid_system_cache", True),
        )


@dataclass
class ModelSettings:
    """Model configuration settings."""

    model_dirs: list[str] = field(default_factory=list)  # [] means ~/.omlx/models
    model_dir: str | None = None  # Deprecated: kept for backward compatibility
    model_fallback: bool = False  # Use default model when requested model not found
    hide_helper_models: bool = (
        False  # Hide dFlash/Assistant/Draft helper models from /v1/models
    )

    def get_model_dirs(self, base_path: Path) -> list[Path]:
        """
        Get the resolved model directory paths.

        Args:
            base_path: Base oMLX directory.

        Returns:
            List of resolved model directory paths.
        """
        if self.model_dirs:
            return [Path(d).expanduser().resolve() for d in self.model_dirs]
        if self.model_dir:
            return [Path(self.model_dir).expanduser().resolve()]
        return [base_path / "models"]

    def get_model_dir(self, base_path: Path) -> Path:
        """
        Get the primary (first) resolved model directory path.

        Args:
            base_path: Base oMLX directory.

        Returns:
            Resolved primary model directory path.
        """
        return self.get_model_dirs(base_path)[0]

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "model_dirs": self.model_dirs,
            "model_dir": self.model_dirs[0] if self.model_dirs else self.model_dir,
            "model_fallback": self.model_fallback,
            "hide_helper_models": self.hide_helper_models,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> ModelSettings:
        """Create from dictionary."""
        model_dirs = data.get("model_dirs", [])
        # Backward compatibility: migrate old model_dir to model_dirs
        if not model_dirs and data.get("model_dir"):
            model_dirs = [data["model_dir"]]
        return cls(
            model_dirs=model_dirs,
            model_dir=data.get("model_dir"),
            model_fallback=data.get("model_fallback", False),
            hide_helper_models=data.get("hide_helper_models", False),
        )


@dataclass
class SchedulerSettings:
    """Scheduler configuration settings."""

    max_concurrent_requests: int = 8
    embedding_batch_size: int = 32
    # When True, long prefills are interleaved with decode steps.
    # Reduces TTFT for concurrent requests at the cost of per-step overhead.
    chunked_prefill: bool = False

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return asdict(self)

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> SchedulerSettings:
        """Create from dictionary."""
        # Backwards compatibility: migrate old keys
        value = data.get("max_concurrent_requests")
        if value is None:
            value = data.get("max_num_seqs")
        if value is None:
            value = data.get("completion_batch_size")
        if value is None:
            value = 8
        embedding_batch_size = data.get("embedding_batch_size", 32)
        return cls(
            max_concurrent_requests=value,
            embedding_batch_size=embedding_batch_size,
            chunked_prefill=bool(data.get("chunked_prefill", False)),
        )


@dataclass
class CacheSettings:
    """Cache configuration settings."""

    enabled: bool = True
    hot_cache_only: bool = False
    ssd_cache_dir: str | None = None  # None means ~/.omlx/cache
    ssd_cache_max_size: str = "auto"  # "auto" means 10% of SSD capacity
    hot_cache_max_size: str = "0"  # "0" = disabled, e.g. "8GB"
    initial_cache_blocks: int = 256  # Starting blocks (grows dynamically)

    def get_ssd_cache_dir(self, base_path: Path) -> Path:
        """
        Get the resolved SSD cache directory path.

        Args:
            base_path: Base oMLX directory.

        Returns:
            Resolved SSD cache directory path.
        """
        if self.ssd_cache_dir:
            return Path(self.ssd_cache_dir).expanduser().resolve()
        return base_path / "cache"

    def get_ssd_cache_max_size_bytes(self, base_path: Path) -> int:
        """
        Get max SSD cache size in bytes.

        Args:
            base_path: Base oMLX directory.

        Returns:
            Max SSD cache size in bytes (10% of SSD if "auto").
        """
        if self.ssd_cache_max_size.lower() == "auto":
            cache_dir = self.get_ssd_cache_dir(base_path)
            return int(get_ssd_capacity(cache_dir) * 0.1)
        return parse_size(self.ssd_cache_max_size)

    def get_hot_cache_max_size_bytes(self) -> int:
        """Get hot cache max size in bytes. 0 means disabled."""
        return parse_size(self.hot_cache_max_size)

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "enabled": self.enabled,
            "hot_cache_only": self.hot_cache_only,
            "ssd_cache_dir": self.ssd_cache_dir,
            "ssd_cache_max_size": self.ssd_cache_max_size,
            "hot_cache_max_size": self.hot_cache_max_size,
            "initial_cache_blocks": self.initial_cache_blocks,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> CacheSettings:
        """Create from dictionary."""
        hot_cache_max_size = data.get("hot_cache_max_size", "0")
        if isinstance(hot_cache_max_size, str) and hot_cache_max_size.lower() == "auto":
            hot_cache_max_size = "0"

        return cls(
            enabled=data.get("enabled", True),
            hot_cache_only=data.get("hot_cache_only", False),
            ssd_cache_dir=data.get("ssd_cache_dir"),
            ssd_cache_max_size=data.get("ssd_cache_max_size", "auto"),
            hot_cache_max_size=hot_cache_max_size,
            initial_cache_blocks=data.get("initial_cache_blocks", 256),
        )


MemoryGuardTier = Literal["safe", "balanced", "aggressive", "custom"]
VALID_MEMORY_GUARD_TIERS: set[str] = {"safe", "balanced", "aggressive", "custom"}


@dataclass
class MemorySettings:
    """Process-level memory enforcement settings."""

    prefill_memory_guard: bool = (
        True  # Memory guard: prefill estimation + generation scheduling defer
    )
    # Tier selects the active-memory reclaim ratio (safe/balanced/aggressive)
    # or, for "custom", lets the user pin the dynamic ceiling to a fixed
    # GB number. See ProcessMemoryEnforcer._get_dynamic_ceiling for the math.
    memory_guard_tier: MemoryGuardTier = "balanced"
    # Only consulted when memory_guard_tier == "custom". GB. 0 = unset.
    memory_guard_custom_ceiling_gb: float = 0.0
    # Two-stage watermark on the ceiling. soft triggers admission pause + LRU eviction,
    # hard triggers in-flight abort. Gap >= 10% absorbs macOS compressed-memory oscillation.
    soft_threshold: float = 0.85
    hard_threshold: float = 0.95
    # Adaptive prefill throttle. When current memory >= hard_cap * safe_zone_ratio
    # the next chunk is sized so its predicted transient stays under the cap.
    # If even prefill_min_chunk_tokens would exceed the cap, the request is
    # aborted via the same cleanup path the hard-limit RuntimeError uses.
    prefill_safe_zone_ratio: float = 0.80
    prefill_min_chunk_tokens: int = 32

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "prefill_memory_guard": self.prefill_memory_guard,
            "memory_guard_tier": self.memory_guard_tier,
            "memory_guard_custom_ceiling_gb": self.memory_guard_custom_ceiling_gb,
            "soft_threshold": self.soft_threshold,
            "hard_threshold": self.hard_threshold,
            "prefill_safe_zone_ratio": self.prefill_safe_zone_ratio,
            "prefill_min_chunk_tokens": self.prefill_min_chunk_tokens,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> MemorySettings:
        """Create from dictionary."""
        tier = str(data.get("memory_guard_tier", "balanced")).lower()
        if tier not in VALID_MEMORY_GUARD_TIERS:
            tier = "balanced"
        return cls(
            prefill_memory_guard=data.get("prefill_memory_guard", True),
            memory_guard_tier=tier,  # type: ignore[arg-type]
            memory_guard_custom_ceiling_gb=float(
                data.get("memory_guard_custom_ceiling_gb", 0.0)
            ),
            soft_threshold=float(data.get("soft_threshold", 0.85)),
            hard_threshold=float(data.get("hard_threshold", 0.95)),
            prefill_safe_zone_ratio=float(data.get("prefill_safe_zone_ratio", 0.80)),
            prefill_min_chunk_tokens=int(data.get("prefill_min_chunk_tokens", 32)),
        )


@dataclass
class ModelIdleTimeoutSettings:
    """Idle timeout settings for automatic model unloading."""

    idle_timeout_seconds: int | None = None

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {"idle_timeout_seconds": self.idle_timeout_seconds}

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> ModelIdleTimeoutSettings:
        """Create from dictionary."""
        return cls(
            idle_timeout_seconds=data.get("idle_timeout_seconds"),
        )


@dataclass
class SubKeyEntry:
    """A sub API key entry for API-only authentication."""

    key: str
    name: str = ""
    created_at: str = ""

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "key": self.key,
            "name": self.name,
            "created_at": self.created_at,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> SubKeyEntry:
        """Create from dictionary."""
        return cls(
            key=data.get("key", ""),
            name=data.get("name", ""),
            created_at=data.get("created_at", ""),
        )


@dataclass
class AuthSettings:
    """Authentication configuration settings."""

    api_key: str | None = None
    secret_key: str | None = None
    skip_api_key_verification: bool = False
    sub_keys: list[SubKeyEntry] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "api_key": self.api_key,
            "secret_key": self.secret_key,
            "skip_api_key_verification": self.skip_api_key_verification,
            "sub_keys": [sk.to_dict() for sk in self.sub_keys],
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> AuthSettings:
        """Create from dictionary."""
        return cls(
            api_key=data.get("api_key"),
            secret_key=data.get("secret_key"),
            skip_api_key_verification=data.get("skip_api_key_verification", False),
            sub_keys=[SubKeyEntry.from_dict(sk) for sk in data.get("sub_keys", [])],
        )


@dataclass
class MCPSettings:
    """MCP (Model Context Protocol) configuration settings."""

    config_path: str | None = None

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {"config_path": self.config_path}

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> MCPSettings:
        """Create from dictionary."""
        return cls(config_path=data.get("config_path"))


@dataclass
class HuggingFaceSettings:
    """HuggingFace Hub configuration settings."""

    endpoint: str = ""  # Empty string = use HF default (https://huggingface.co)
    hf_cache_enabled: bool = True

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "endpoint": self.endpoint,
            "hf_cache_enabled": self.hf_cache_enabled,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> HuggingFaceSettings:
        """Create from dictionary."""
        return cls(
            endpoint=data.get("endpoint", ""),
            hf_cache_enabled=data.get("hf_cache_enabled", True),
        )


@dataclass
class ModelScopeSettings:
    """ModelScope Hub configuration settings."""

    endpoint: str = ""  # Empty string = use default (https://modelscope.cn)

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {"endpoint": self.endpoint}

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> ModelScopeSettings:
        """Create from dictionary."""
        return cls(endpoint=data.get("endpoint", ""))


@dataclass
class NetworkSettings:
    """Network proxy and TLS trust settings."""

    http_proxy: str = ""
    https_proxy: str = ""
    no_proxy: str = ""
    ca_bundle: str = ""

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "http_proxy": self.http_proxy,
            "https_proxy": self.https_proxy,
            "no_proxy": self.no_proxy,
            "ca_bundle": self.ca_bundle,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> NetworkSettings:
        """Create from dictionary."""
        return cls(
            http_proxy=data.get("http_proxy", ""),
            https_proxy=data.get("https_proxy", ""),
            no_proxy=data.get("no_proxy", ""),
            ca_bundle=data.get("ca_bundle", ""),
        )


@dataclass
class SamplingSettings:
    """Default sampling parameters for generation."""

    # Fallback context length used by ``server.get_max_context_window``
    # only when neither a per-model override nor a model-config
    # discovered native context length is available. Default kept at
    # 32768 so existing ``settings.json`` files carrying the historical
    # default keep working unchanged after upgrade.
    max_context_window: int = 32768
    # Optional operator policy cap. When set, the server returns
    # ``min(native_context, max_context_window_policy)`` for models
    # whose native context length is discovered. Unset (None) by
    # default so no install behavior changes implicitly. Per-model
    # overrides and the fallback default above are not affected.
    max_context_window_policy: int | None = None
    max_tokens: int = 32768
    temperature: float = 1.0
    top_p: float = 0.95
    top_k: int = 0
    repetition_penalty: float = 1.0

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "max_context_window": self.max_context_window,
            "max_context_window_policy": self.max_context_window_policy,
            "max_tokens": self.max_tokens,
            "temperature": self.temperature,
            "top_p": self.top_p,
            "top_k": self.top_k,
            "repetition_penalty": self.repetition_penalty,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> SamplingSettings:
        """Create from dictionary."""
        return cls(
            max_context_window=data.get("max_context_window", 32768),
            max_context_window_policy=data.get("max_context_window_policy"),
            max_tokens=data.get("max_tokens", 32768),
            temperature=data.get("temperature", 1.0),
            top_p=data.get("top_p", 0.95),
            top_k=data.get("top_k", 0),
            repetition_penalty=data.get("repetition_penalty", 1.0),
        )


@dataclass
class LoggingSettings:
    """Logging configuration settings."""

    log_dir: str | None = None  # None means {base_path}/logs
    retention_days: int = 7  # Number of days to keep rotated log files

    def get_log_dir(self, base_path: Path) -> Path:
        """
        Get the resolved log directory path.

        Args:
            base_path: Base oMLX directory.

        Returns:
            Resolved log directory path.
        """
        if self.log_dir:
            return Path(self.log_dir).expanduser().resolve()
        return base_path / "logs"

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "log_dir": self.log_dir,
            "retention_days": self.retention_days,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> LoggingSettings:
        """Create from dictionary."""
        return cls(
            log_dir=data.get("log_dir"),
            retention_days=data.get("retention_days", 7),
        )


@dataclass
class UISettings:
    """Admin UI settings."""

    language: str = "en"

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {"language": self.language}

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> UISettings:
        """Create from dictionary."""
        return cls(language=data.get("language", "en"))


@dataclass
class ClaudeCodeSettings:
    """Claude Code integration settings."""

    context_scaling_enabled: bool = False
    target_context_size: int = 200000  # Claude Code default (200k)
    # Mode: "cloud" = native claude.ai subscription, "local" = route through omlx.
    # Default is "cloud" so upgrades don't silently route traffic to omlx.
    mode: str = "cloud"
    opus_model: str | None = None
    sonnet_model: str | None = None
    haiku_model: str | None = None

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "context_scaling_enabled": self.context_scaling_enabled,
            "target_context_size": self.target_context_size,
            "mode": self.mode,
            "opus_model": self.opus_model,
            "sonnet_model": self.sonnet_model,
            "haiku_model": self.haiku_model,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> ClaudeCodeSettings:
        """Create from dictionary."""
        return cls(
            context_scaling_enabled=data.get("context_scaling_enabled", False),
            target_context_size=data.get("target_context_size", 200000),
            mode=data.get("mode", "cloud"),
            opus_model=data.get("opus_model"),
            sonnet_model=data.get("sonnet_model"),
            haiku_model=data.get("haiku_model"),
        )


@dataclass
class IntegrationSettings:
    """Other integrations settings."""

    codex_model: str | None = None
    opencode_model: str | None = None
    openclaw_model: str | None = None
    hermes_model: str | None = None
    pi_model: str | None = None
    copilot_model: str | None = None
    openclaw_tools_profile: str = "coding"
    markitdown_enabled: bool = True
    markitdown_expose_model: bool = False
    markitdown_max_file_size_mb: int = 25
    markitdown_max_files_per_request: int = 5
    markitdown_pdf_processing_engine: str = "markitdown"

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "codex_model": self.codex_model,
            "opencode_model": self.opencode_model,
            "openclaw_model": self.openclaw_model,
            "hermes_model": self.hermes_model,
            "pi_model": self.pi_model,
            "copilot_model": self.copilot_model,
            "openclaw_tools_profile": self.openclaw_tools_profile,
            "markitdown_enabled": self.markitdown_enabled,
            "markitdown_expose_model": self.markitdown_expose_model,
            "markitdown_max_file_size_mb": self.markitdown_max_file_size_mb,
            "markitdown_max_files_per_request": self.markitdown_max_files_per_request,
            "markitdown_pdf_processing_engine": self.markitdown_pdf_processing_engine,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> IntegrationSettings:
        """Create from dictionary."""
        return cls(
            codex_model=data.get("codex_model"),
            opencode_model=data.get("opencode_model"),
            openclaw_model=data.get("openclaw_model"),
            hermes_model=data.get("hermes_model"),
            pi_model=data.get("pi_model"),
            copilot_model=data.get("copilot_model"),
            openclaw_tools_profile=data.get("openclaw_tools_profile", "coding"),
            markitdown_enabled=data.get("markitdown_enabled", True),
            markitdown_expose_model=data.get("markitdown_expose_model", False),
            markitdown_max_file_size_mb=data.get("markitdown_max_file_size_mb", 25),
            markitdown_max_files_per_request=data.get(
                "markitdown_max_files_per_request", 5
            ),
            markitdown_pdf_processing_engine=data.get(
                "markitdown_pdf_processing_engine", "markitdown"
            ),
        )


@dataclass
class GlobalSettings:
    """
    Global settings for oMLX.

    Combines all settings sections and provides methods for:
    - Loading from file with CLI/env overrides
    - Saving to file
    - Directory management
    - Validation
    """

    base_path: Path = field(default_factory=lambda: DEFAULT_BASE_PATH)
    server: ServerSettings = field(default_factory=ServerSettings)
    model: ModelSettings = field(default_factory=ModelSettings)
    memory: MemorySettings = field(default_factory=MemorySettings)
    scheduler: SchedulerSettings = field(default_factory=SchedulerSettings)
    cache: CacheSettings = field(default_factory=CacheSettings)
    auth: AuthSettings = field(default_factory=AuthSettings)
    mcp: MCPSettings = field(default_factory=MCPSettings)
    huggingface: HuggingFaceSettings = field(default_factory=HuggingFaceSettings)
    modelscope: ModelScopeSettings = field(default_factory=ModelScopeSettings)
    network: NetworkSettings = field(default_factory=NetworkSettings)
    sampling: SamplingSettings = field(default_factory=SamplingSettings)
    logging: LoggingSettings = field(default_factory=LoggingSettings)
    claude_code: ClaudeCodeSettings = field(default_factory=ClaudeCodeSettings)
    integrations: IntegrationSettings = field(default_factory=IntegrationSettings)
    ui: UISettings = field(default_factory=UISettings)
    idle_timeout: ModelIdleTimeoutSettings = field(
        default_factory=ModelIdleTimeoutSettings
    )

    @classmethod
    def load(
        cls,
        base_path: str | Path | None = None,
        cli_args: Any | None = None,
    ) -> GlobalSettings:
        """
        Load settings with priority hierarchy: CLI > env > file > defaults.

        Args:
            base_path: Base directory for oMLX (default: resolved via
                OMLX_BASE_PATH env var, the macOS app's bootstrap file,
                then ~/.omlx).
            cli_args: Argparse namespace with CLI arguments.

        Returns:
            Loaded GlobalSettings instance.
        """
        # Resolve base path
        if base_path:
            resolved_base = Path(base_path).expanduser().resolve()
        else:
            resolved_base = resolve_default_base_path()

        # Start with defaults
        settings = cls(base_path=resolved_base)

        # Load from file if exists
        settings_file = resolved_base / "settings.json"
        if settings_file.exists():
            settings._load_from_file(settings_file)
            logger.debug(f"Loaded settings from {settings_file}")

        # Apply environment variable overrides
        settings._apply_env_overrides()

        # Apply CLI argument overrides
        if cli_args:
            settings._apply_cli_overrides(cli_args)

        return settings

    def _load_from_file(self, path: Path) -> None:
        """
        Load settings from a JSON file.

        Args:
            path: Path to the settings JSON file.
        """
        try:
            with open(path, encoding="utf-8") as f:
                data = json.load(f)

            # Check version for future migrations
            version = data.get("version", "1.0")
            if version != SETTINGS_VERSION:
                logger.info(
                    f"Settings file version {version} differs from "
                    f"current {SETTINGS_VERSION}, migrating..."
                )

            # Load each section
            if "server" in data:
                self.server = ServerSettings.from_dict(data["server"])
            if "model" in data:
                self.model = ModelSettings.from_dict(data["model"])
            if "memory" in data:
                self.memory = MemorySettings.from_dict(data["memory"])
            if "scheduler" in data:
                self.scheduler = SchedulerSettings.from_dict(data["scheduler"])
            if "cache" in data:
                self.cache = CacheSettings.from_dict(data["cache"])
            if "auth" in data:
                self.auth = AuthSettings.from_dict(data["auth"])
            if "mcp" in data:
                self.mcp = MCPSettings.from_dict(data["mcp"])
            if "huggingface" in data:
                self.huggingface = HuggingFaceSettings.from_dict(data["huggingface"])
            if "modelscope" in data:
                self.modelscope = ModelScopeSettings.from_dict(data["modelscope"])
            if "network" in data:
                self.network = NetworkSettings.from_dict(data["network"])
            if "sampling" in data:
                self.sampling = SamplingSettings.from_dict(data["sampling"])
            if "logging" in data:
                self.logging = LoggingSettings.from_dict(data["logging"])
            if "claude_code" in data:
                self.claude_code = ClaudeCodeSettings.from_dict(data["claude_code"])
            if "integrations" in data:
                self.integrations = IntegrationSettings.from_dict(data["integrations"])
            if "ui" in data:
                self.ui = UISettings.from_dict(data["ui"])
            if "idle_timeout" in data:
                self.idle_timeout = ModelIdleTimeoutSettings.from_dict(
                    data["idle_timeout"]
                )

        except json.JSONDecodeError as e:
            logger.warning(f"Failed to parse settings file {path}: {e}")
        except OSError as e:
            logger.warning(f"Failed to read settings file {path}: {e}")

    def _apply_env_overrides(self) -> None:
        """Apply OMLX_* environment variable overrides."""
        # Server settings
        if host := os.getenv("OMLX_HOST"):
            self.server.host = host
        if port := os.getenv("OMLX_PORT"):
            try:
                self.server.port = int(port)
            except ValueError:
                logger.warning(f"Invalid OMLX_PORT value: {port}")
        if log_level := os.getenv("OMLX_LOG_LEVEL"):
            self.server.log_level = log_level
        if preserve_mid_system_cache := os.getenv("OMLX_PRESERVE_MID_SYSTEM_CACHE"):
            self.server.preserve_mid_system_cache = (
                preserve_mid_system_cache.strip().lower() in {"1", "true", "yes", "on"}
            )

        # Model settings
        if model_dir := os.getenv("OMLX_MODEL_DIR"):
            dirs = [d.strip() for d in model_dir.split(",") if d.strip()]
            self.model.model_dirs = dirs
            self.model.model_dir = dirs[0] if dirs else None
        # Scheduler settings
        max_concurrent = os.getenv("OMLX_MAX_CONCURRENT_REQUESTS") or os.getenv(
            "OMLX_MAX_NUM_SEQS"
        )
        if max_concurrent:
            try:
                self.scheduler.max_concurrent_requests = int(max_concurrent)
            except ValueError:
                logger.warning(
                    f"Invalid OMLX_MAX_CONCURRENT_REQUESTS value: {max_concurrent}"
                )
        if embedding_batch_size := os.getenv("OMLX_EMBEDDING_BATCH_SIZE"):
            try:
                self.scheduler.embedding_batch_size = int(embedding_batch_size)
            except ValueError:
                logger.warning(
                    f"Invalid OMLX_EMBEDDING_BATCH_SIZE value: {embedding_batch_size}"
                )

        # Cache settings
        if cache_enabled := os.getenv("OMLX_CACHE_ENABLED"):
            self.cache.enabled = cache_enabled.lower() in ("true", "1", "yes")
        if ssd_cache_dir := os.getenv("OMLX_SSD_CACHE_DIR"):
            self.cache.ssd_cache_dir = ssd_cache_dir
        if ssd_cache_max := os.getenv("OMLX_SSD_CACHE_MAX_SIZE"):
            self.cache.ssd_cache_max_size = ssd_cache_max
        if hot_cache_only := os.getenv("OMLX_HOT_CACHE_ONLY"):
            self.cache.hot_cache_only = hot_cache_only.lower() in ("true", "1", "yes")
        if initial_blocks := os.getenv("OMLX_INITIAL_CACHE_BLOCKS"):
            try:
                self.cache.initial_cache_blocks = int(initial_blocks)
            except ValueError:
                logger.warning(
                    f"Invalid OMLX_INITIAL_CACHE_BLOCKS value: {initial_blocks}"
                )

        # Auth settings
        if api_key := os.getenv("OMLX_API_KEY"):
            self.auth.api_key = api_key

        # MCP settings
        if mcp_config := os.getenv("OMLX_MCP_CONFIG"):
            self.mcp.config_path = mcp_config

        # HuggingFace settings
        if hf_endpoint := os.getenv("OMLX_HF_ENDPOINT"):
            self.huggingface.endpoint = hf_endpoint
        if hf_cache_enabled := os.getenv("OMLX_HF_CACHE_ENABLED"):
            self.huggingface.hf_cache_enabled = hf_cache_enabled.strip().lower() in {
                "1",
                "true",
                "yes",
                "on",
            }

        # ModelScope settings
        if ms_endpoint := os.getenv("OMLX_MS_ENDPOINT"):
            self.modelscope.endpoint = ms_endpoint

        # Network settings
        if http_proxy := os.getenv("OMLX_HTTP_PROXY"):
            self.network.http_proxy = http_proxy
        if https_proxy := os.getenv("OMLX_HTTPS_PROXY"):
            self.network.https_proxy = https_proxy
        if no_proxy := os.getenv("OMLX_NO_PROXY"):
            self.network.no_proxy = no_proxy
        if ca_bundle := os.getenv("OMLX_CA_BUNDLE"):
            self.network.ca_bundle = ca_bundle

        # Logging settings
        if log_dir := os.getenv("OMLX_LOG_DIR"):
            self.logging.log_dir = log_dir
        if retention_days := os.getenv("OMLX_LOG_RETENTION_DAYS"):
            try:
                self.logging.retention_days = int(retention_days)
            except ValueError:
                logger.warning(f"Invalid OMLX_LOG_RETENTION_DAYS: {retention_days}")

        # Integration settings
        if markitdown_enabled := os.getenv("OMLX_MARKITDOWN_ENABLED"):
            self.integrations.markitdown_enabled = (
                markitdown_enabled.strip().lower() in {"1", "true", "yes", "on"}
            )
        if markitdown_expose_model := os.getenv("OMLX_MARKITDOWN_EXPOSE_MODEL"):
            self.integrations.markitdown_expose_model = (
                markitdown_expose_model.strip().lower() in {"1", "true", "yes", "on"}
            )
        if markitdown_pdf_processing_engine := os.getenv(
            "OMLX_MARKITDOWN_PDF_PROCESSING_ENGINE"
        ):
            self.integrations.markitdown_pdf_processing_engine = (
                markitdown_pdf_processing_engine.strip() or "markitdown"
            )

    def _apply_cli_overrides(self, args: Any) -> None:
        """
        Apply CLI argument overrides.

        Args:
            args: Argparse namespace with CLI arguments.
        """
        # Server settings
        if hasattr(args, "host") and args.host is not None:
            self.server.host = args.host
        if hasattr(args, "port") and args.port is not None:
            self.server.port = args.port
        if hasattr(args, "log_level") and args.log_level is not None:
            self.server.log_level = args.log_level
        if hasattr(args, "sse_keepalive_mode") and args.sse_keepalive_mode is not None:
            self.server.sse_keepalive_mode = args.sse_keepalive_mode

        # Model settings
        if hasattr(args, "model_dir") and args.model_dir is not None:
            dirs = [d.strip() for d in args.model_dir.split(",") if d.strip()]
            self.model.model_dirs = dirs
            self.model.model_dir = dirs[0] if dirs else None
        # Scheduler settings
        if (
            hasattr(args, "max_concurrent_requests")
            and args.max_concurrent_requests is not None
        ):
            self.scheduler.max_concurrent_requests = args.max_concurrent_requests
        if (
            hasattr(args, "embedding_batch_size")
            and args.embedding_batch_size is not None
        ):
            self.scheduler.embedding_batch_size = args.embedding_batch_size

        # Memory guard settings
        if hasattr(args, "memory_guard") and args.memory_guard is not None:
            self.memory.memory_guard_tier = args.memory_guard
        if hasattr(args, "memory_guard_gb") and args.memory_guard_gb is not None:
            self.memory.memory_guard_tier = "custom"
            self.memory.memory_guard_custom_ceiling_gb = float(args.memory_guard_gb)

        # Cache settings
        if hasattr(args, "cache_enabled") and args.cache_enabled is not None:
            self.cache.enabled = args.cache_enabled
        if hasattr(args, "ssd_cache_dir") and args.ssd_cache_dir is not None:
            self.cache.ssd_cache_dir = args.ssd_cache_dir
        if hasattr(args, "ssd_cache_max_size") and args.ssd_cache_max_size is not None:
            self.cache.ssd_cache_max_size = args.ssd_cache_max_size
        if (
            hasattr(args, "initial_cache_blocks")
            and args.initial_cache_blocks is not None
        ):
            self.cache.initial_cache_blocks = args.initial_cache_blocks

        # Auth settings
        if hasattr(args, "api_key") and args.api_key is not None:
            self.auth.api_key = args.api_key

        # MCP settings
        if hasattr(args, "mcp_config") and args.mcp_config is not None:
            self.mcp.config_path = args.mcp_config

        # HuggingFace settings
        if hasattr(args, "hf_endpoint") and args.hf_endpoint is not None:
            self.huggingface.endpoint = args.hf_endpoint
        if hasattr(args, "hf_cache_enabled") and args.hf_cache_enabled is not None:
            self.huggingface.hf_cache_enabled = args.hf_cache_enabled

        # ModelScope settings
        if hasattr(args, "ms_endpoint") and args.ms_endpoint is not None:
            self.modelscope.endpoint = args.ms_endpoint

        # Network settings
        if hasattr(args, "http_proxy") and args.http_proxy is not None:
            self.network.http_proxy = args.http_proxy
        if hasattr(args, "https_proxy") and args.https_proxy is not None:
            self.network.https_proxy = args.https_proxy
        if hasattr(args, "no_proxy") and args.no_proxy is not None:
            self.network.no_proxy = args.no_proxy
        if hasattr(args, "ca_bundle") and args.ca_bundle is not None:
            self.network.ca_bundle = args.ca_bundle

    def get_hf_cache_dir(self) -> Path:
        """Return the standard HuggingFace Hub cache directory."""
        if hf_hub_cache := os.getenv("HF_HUB_CACHE"):
            return Path(hf_hub_cache).expanduser().resolve()
        if hf_home := os.getenv("HF_HOME"):
            return (Path(hf_home).expanduser() / "hub").resolve()
        return (Path.home() / ".cache" / "huggingface" / "hub").resolve()

    def get_effective_model_dirs(
        self, model_dirs: list[str] | None = None
    ) -> list[Path]:
        """Return model directories in discovery order, including HF cache."""
        if model_dirs is None:
            configured = self.model.get_model_dirs(self.base_path)
        elif model_dirs:
            configured = [Path(d).expanduser().resolve() for d in model_dirs]
        else:
            configured = [self.base_path / "models"]
        effective: list[Path] = []
        seen: set[Path] = set()

        def add(path: Path, *, require_exists: bool = False) -> None:
            resolved = path.expanduser().resolve()
            if require_exists and not resolved.exists():
                return
            if resolved in seen:
                return
            seen.add(resolved)
            effective.append(resolved)

        if configured:
            add(configured[0])
        if self.huggingface.hf_cache_enabled:
            add(self.get_hf_cache_dir(), require_exists=True)
        for directory in configured[1:]:
            add(directory)

        return effective

    def save(self) -> None:
        """Save current settings to the settings file."""
        self.ensure_directories()

        settings_file = self.base_path / "settings.json"
        data = {
            "version": SETTINGS_VERSION,
            "server": self.server.to_dict(),
            "model": self.model.to_dict(),
            "memory": self.memory.to_dict(),
            "scheduler": self.scheduler.to_dict(),
            "cache": self.cache.to_dict(),
            "auth": self.auth.to_dict(),
            "mcp": self.mcp.to_dict(),
            "huggingface": self.huggingface.to_dict(),
            "modelscope": self.modelscope.to_dict(),
            "network": self.network.to_dict(),
            "sampling": self.sampling.to_dict(),
            "logging": self.logging.to_dict(),
            "claude_code": self.claude_code.to_dict(),
            "integrations": self.integrations.to_dict(),
            "ui": self.ui.to_dict(),
            "idle_timeout": self.idle_timeout.to_dict(),
        }

        try:
            with open(settings_file, "w", encoding="utf-8") as f:
                json.dump(data, f, indent=2)
            logger.info(f"Saved settings to {settings_file}")
        except OSError as e:
            logger.error(f"Failed to save settings to {settings_file}: {e}")
            raise

    def ensure_directories(self) -> None:
        """Create necessary directories if they don't exist."""
        from .model_discovery import model_directory_access_error

        # Required directories - fatal if creation fails
        required = [
            self.base_path,
            self.cache.get_ssd_cache_dir(self.base_path),
            self.logging.get_log_dir(self.base_path),
        ]

        for directory in required:
            if not directory.exists():
                try:
                    directory.mkdir(parents=True, exist_ok=True)
                    logger.debug(f"Created directory: {directory}")
                except OSError as e:
                    logger.error(f"Failed to create directory {directory}: {e}")
                    raise

        # Model directories - skip unavailable paths (e.g. disconnected external drive)
        valid_dirs = []
        for directory in self.model.get_model_dirs(self.base_path):
            if not directory.exists():
                try:
                    directory.mkdir(parents=True, exist_ok=True)
                    logger.debug(f"Created directory: {directory}")
                except OSError as e:
                    logger.warning(
                        f"Model directory unavailable, skipping: {directory} ({e})"
                    )
                    continue

            access_error = model_directory_access_error(directory)
            if access_error is not None:
                logger.warning(f"Model directory unavailable, skipping: {access_error}")
                continue

            valid_dirs.append(str(directory))

        # Update model_dirs to only include valid paths
        self.model.model_dirs = valid_dirs
        self.model.model_dir = None

    def validate(self) -> list[str]:
        """
        Validate all settings.

        Returns:
            List of validation error messages (empty if valid).
        """
        errors = []

        # Server validation
        if not 1 <= self.server.port <= 65535:
            errors.append(f"Invalid port: {self.server.port} (must be 1-65535)")

        valid_log_levels = {"trace", "debug", "info", "warning", "error", "critical"}
        if self.server.log_level.lower() not in valid_log_levels:
            errors.append(
                f"Invalid log_level: {self.server.log_level} "
                f"(must be one of {valid_log_levels})"
            )

        valid_keepalive_modes = {"chunk", "comment", "off"}
        if self.server.sse_keepalive_mode not in valid_keepalive_modes:
            errors.append(
                f"Invalid sse_keepalive_mode: {self.server.sse_keepalive_mode} "
                f"(must be one of {valid_keepalive_modes})"
            )

        # Memory guard tier validation
        if self.memory.memory_guard_tier not in VALID_MEMORY_GUARD_TIERS:
            errors.append(
                f"Invalid memory_guard_tier: {self.memory.memory_guard_tier} "
                f"(must be one of {sorted(VALID_MEMORY_GUARD_TIERS)})"
            )

        # Custom ceiling must be > 0 when tier == "custom"
        if (
            self.memory.memory_guard_tier == "custom"
            and self.memory.memory_guard_custom_ceiling_gb <= 0
        ):
            errors.append(
                "memory_guard_custom_ceiling_gb must be > 0 when "
                "memory_guard_tier is 'custom'"
            )

        if not 0.5 <= self.memory.prefill_safe_zone_ratio <= 0.99:
            errors.append(
                f"prefill_safe_zone_ratio must be in [0.5, 0.99], "
                f"got {self.memory.prefill_safe_zone_ratio}"
            )
        if not 1 <= self.memory.prefill_min_chunk_tokens <= 1024:
            errors.append(
                f"prefill_min_chunk_tokens must be in [1, 1024], "
                f"got {self.memory.prefill_min_chunk_tokens}"
            )

        # Scheduler validation
        if self.scheduler.max_concurrent_requests <= 0:
            errors.append(
                f"Invalid max_concurrent_requests: "
                f"{self.scheduler.max_concurrent_requests} (must be > 0)"
            )
        if self.scheduler.embedding_batch_size <= 0:
            errors.append(
                f"Invalid embedding_batch_size: "
                f"{self.scheduler.embedding_batch_size} (must be > 0)"
            )

        # Cache validation
        if self.cache.ssd_cache_max_size.lower() != "auto":
            try:
                size = parse_size(self.cache.ssd_cache_max_size)
                if size <= 0:
                    errors.append("ssd_cache_max_size must be positive")
            except ValueError as e:
                errors.append(f"Invalid ssd_cache_max_size: {e}")

        try:
            hot_cache_size = parse_size(self.cache.hot_cache_max_size)
            if hot_cache_size < 0:
                errors.append("hot_cache_max_size must be non-negative")
        except ValueError as e:
            if self.cache.hot_cache_max_size.strip().lower() == "auto":
                errors.append(
                    "Invalid hot_cache_max_size: 'auto' is not supported; "
                    "use '0' to disable or a size like '8GB'"
                )
            else:
                errors.append(f"Invalid hot_cache_max_size: {e}")

        if self.cache.initial_cache_blocks <= 0:
            errors.append(
                f"Invalid initial_cache_blocks: "
                f"{self.cache.initial_cache_blocks} (must be > 0)"
            )

        # Sampling validation
        if (
            self.sampling.max_context_window_policy is not None
            and self.sampling.max_context_window_policy <= 0
        ):
            errors.append(
                "Invalid sampling max_context_window_policy: "
                f"{self.sampling.max_context_window_policy} (must be > 0)"
            )
        if self.sampling.max_tokens <= 0:
            errors.append(
                f"Invalid sampling max_tokens: {self.sampling.max_tokens} (must be > 0)"
            )
        if not 0.0 <= self.sampling.temperature <= 2.0:
            errors.append(
                f"Invalid sampling temperature: {self.sampling.temperature} "
                "(must be 0.0-2.0)"
            )
        if not 0.0 <= self.sampling.top_p <= 1.0:
            errors.append(
                f"Invalid sampling top_p: {self.sampling.top_p} (must be 0.0-1.0)"
            )
        if self.sampling.top_k < 0:
            errors.append(
                f"Invalid sampling top_k: {self.sampling.top_k} (must be >= 0)"
            )

        # Claude Code validation
        if self.claude_code.target_context_size <= 0:
            errors.append(
                f"Invalid target_context_size: "
                f"{self.claude_code.target_context_size} (must be > 0)"
            )
        valid_modes = {"local", "cloud"}
        if self.claude_code.mode not in valid_modes:
            errors.append(
                f"Invalid claude_code mode: '{self.claude_code.mode}' "
                f"(must be one of {sorted(valid_modes)})"
            )

        # Integration validation
        if self.integrations.markitdown_max_file_size_mb <= 0:
            errors.append("markitdown_max_file_size_mb must be > 0")
        if self.integrations.markitdown_max_files_per_request <= 0:
            errors.append("markitdown_max_files_per_request must be > 0")
        if not str(self.integrations.markitdown_pdf_processing_engine or "").strip():
            errors.append("markitdown_pdf_processing_engine must not be empty")

        # HuggingFace validation
        if self.huggingface.endpoint:
            endpoint = self.huggingface.endpoint.strip()
            if endpoint and not endpoint.startswith(("http://", "https://")):
                errors.append(
                    f"Invalid huggingface endpoint: '{endpoint}' "
                    "(must start with http:// or https://)"
                )

        # ModelScope validation
        if self.modelscope.endpoint:
            endpoint = self.modelscope.endpoint.strip()
            if endpoint and not endpoint.startswith(("http://", "https://")):
                errors.append(
                    f"Invalid modelscope endpoint: '{endpoint}' "
                    "(must start with http:// or https://)"
                )

        # Network proxy validation
        if self.network.http_proxy:
            proxy = self.network.http_proxy.strip()
            if proxy and not proxy.startswith(("http://", "https://")):
                errors.append(
                    f"Invalid http_proxy: '{proxy}' "
                    "(must start with http:// or https://)"
                )
        if self.network.https_proxy:
            proxy = self.network.https_proxy.strip()
            if proxy and not proxy.startswith(("http://", "https://")):
                errors.append(
                    f"Invalid https_proxy: '{proxy}' "
                    "(must start with http:// or https://)"
                )

        return errors

    def to_scheduler_config(self) -> SchedulerConfig:
        """
        Convert settings to SchedulerConfig for engine initialization.

        Returns:
            SchedulerConfig instance with values from settings.
        """
        from .scheduler import SchedulerConfig

        # Always resolve ssd_dir so the scheduler can initialize PagedSSDCacheManager.
        # When hot_cache_only=True, PagedSSDCacheManager skips directory init and
        # the writer thread internally — the dir is not used for disk I/O.
        ssd_dir = (
            self.cache.get_ssd_cache_dir(self.base_path) if self.cache.enabled else None
        )

        return SchedulerConfig(
            max_num_seqs=self.scheduler.max_concurrent_requests,
            completion_batch_size=self.scheduler.max_concurrent_requests,
            embedding_batch_size=self.scheduler.embedding_batch_size,
            chunked_prefill=self.scheduler.chunked_prefill,
            initial_cache_blocks=self.cache.initial_cache_blocks,
            paged_ssd_cache_dir=str(ssd_dir) if ssd_dir else None,
            hot_cache_only=self.cache.hot_cache_only,
            paged_ssd_cache_max_size=self.cache.get_ssd_cache_max_size_bytes(
                self.base_path
            ),
            hot_cache_max_size=self.cache.get_hot_cache_max_size_bytes(),
        )

    def to_dict(self) -> dict[str, Any]:
        """Convert all settings to a dictionary."""
        return {
            "version": SETTINGS_VERSION,
            "base_path": str(self.base_path),
            "server": self.server.to_dict(),
            "model": self.model.to_dict(),
            "memory": self.memory.to_dict(),
            "scheduler": self.scheduler.to_dict(),
            "cache": self.cache.to_dict(),
            "auth": self.auth.to_dict(),
            "mcp": self.mcp.to_dict(),
            "huggingface": self.huggingface.to_dict(),
            "modelscope": self.modelscope.to_dict(),
            "network": self.network.to_dict(),
            "sampling": self.sampling.to_dict(),
            "logging": self.logging.to_dict(),
            "claude_code": self.claude_code.to_dict(),
            "integrations": self.integrations.to_dict(),
            "ui": self.ui.to_dict(),
            "idle_timeout": self.idle_timeout.to_dict(),
        }


# Global singleton instance
_global_settings: GlobalSettings | None = None


def get_settings() -> GlobalSettings:
    """
    Get the global settings instance.

    Returns:
        The global GlobalSettings instance.

    Raises:
        RuntimeError: If settings have not been initialized.
    """
    global _global_settings
    if _global_settings is None:
        raise RuntimeError("Settings not initialized. Call init_settings() first.")
    return _global_settings


def init_settings(
    base_path: str | Path | None = None,
    cli_args: Any | None = None,
) -> GlobalSettings:
    """
    Initialize global settings (call once at startup).

    Args:
        base_path: Base directory for oMLX (default: resolved via
                OMLX_BASE_PATH env var, the macOS app's bootstrap file,
                then ~/.omlx).
        cli_args: Argparse namespace with CLI arguments.

    Returns:
        The initialized GlobalSettings instance.
    """
    global _global_settings
    _global_settings = GlobalSettings.load(base_path=base_path, cli_args=cli_args)
    logger.info(f"Initialized settings with base_path: {_global_settings.base_path}")
    return _global_settings


def reset_settings() -> None:
    """
    Reset global settings (primarily for testing).

    This clears the global singleton, allowing init_settings to be called again.
    """
    global _global_settings
    _global_settings = None