Engines¶

@classmethod
def begin_memory_window(cls) -> Any:
    """Open a measurement window on the loaded child for one turn; the
    returned token closes it (``end_memory_window``). No-op here."""
    return None

@classmethod
def brake_point(cls, point: str, *, llm_id: Any = None) -> BrakeOutcome:
    """Take the brake's sample at a safe point (``turn_start`` for
    ``llm_id``, ``turn_end``, ``idle``) and act on it. No-op here."""
    return BrakeOutcome()

@classmethod
def brake_reload_pending(cls, llm_id: Any) -> bool:
    """Whether the next turn for ``llm_id`` reloads a model the brake
    unloaded (or, on the CPU engine, re-prefills a child it respawned).
    Never here."""
    return False

@classmethod
def chat_template_caps(cls, llm_id: Any = None) -> dict:
    """What the CURRENTLY LOADED child reported about its chat template.

    llama.cpp engines stamp the server's ``chat_template_caps`` map on the
    handle during the one ``/props`` read at spawn -- nine booleans
    evaluated by llama.cpp's own symbolic execution of the template, of
    which ``supports_reasoning_effort`` drives the reasoning-effort lever.
    Engines whose server reports nothing (MLX) return an empty map, and so
    does a child that is down or serving another model.

    ``llm_id`` guards exactly that: capabilities belong to the artifact the
    child was spawned for, so a caller planning a turn for a DIFFERENT
    model gets nothing rather than the loaded model's answers.
    """
    if llm_id is not None and cls._model_id != llm_id:
        return {}
    model = cls._model
    if isinstance(model, dict):
        caps = model.get("chat_template_caps")
        if isinstance(caps, dict):
            return caps
    return {}

@classmethod
def claim_prefix(cls, owner: str) -> None:
    """Make the loaded child's prefix cache belong to ``owner``
    (``conv:<thread id>`` or ``arena``), resetting it when it held
    something else. No-op here."""
    return None

@classmethod
def cleanup(cls) -> None:
    """Free model and tokenizer from memory, reset state.

    Releases GPU/CPU memory occupied by the model and tokenizer,
    resets all cached state. Called automatically after idle timeout
    or manually when switching models.

    Note:
        Thread-safe. Can be called explicitly or by cleanup monitor.
        Calls garbage collector to ensure memory is freed.

    """
    cls._idle_unload = None
    if cls._model or cls._tokenizer:
        import gc

        logger.info(f"Cleaning up model {cls._model_id}")
        cls._model = cls._tokenizer = cls._model_id = cls._last_used = None
        gc.collect()

@classmethod
def community_search_kwargs(cls, term: str) -> dict:
    """`HfApi.list_models` kwargs to find models in THIS engine's runnable
    format across ALL of HF (any author), so community fine-tunes surface and
    are runnable by construction. Searches by the format TAG, not by org —
    e.g. ``filter="mlx"`` catches non-mlx-community quants the org never holds.
    An empty ``term`` is the global pass: filter by format only, ranked by
    downloads.

    ``expand`` pulls pipeline_tag + tags (alongside the downloads/likes the
    quality floor needs) in the one call, so the non-chat task filter (#242)
    can gate ASR/embedding/OCR/media-gen repos out of the derived catalog,
    plus ``gated`` so a repo the token-less app cannot download is dropped
    (``model_resolver.is_gated``).
    """
    kwargs = {
        "filter": cls.FORMAT_TAG,
        "expand": ["pipeline_tag", "tags", "downloads", "likes", "gated"],
    }
    if term:
        kwargs["search"] = term
    return kwargs

@classmethod
def compute_supports_tools(cls, local_path: Union[str, Path]) -> bool:
    """Static tool-calling capability of the model at ``local_path``.

    Computed once at post-download time from the model's chat template by
    differential rendering (see ``engines.tool_capability``): no inference,
    no runtime probe. Any failure (unreadable artifact, missing template)
    returns ``False`` so the model routes through the systematic KB path —
    the capability is never assumed.
    """
    from src.engines.tool_capability import tokenizer_declares_tools

    try:
        tokenizer = cls._load_capability_tokenizer(local_path)
    except Exception:
        # exc_info matters: a missing dependency in the frozen build (the
        # gguf package, #171) is indistinguishable from an unreadable model
        # without the underlying exception in the log.
        logger.warning(
            f"[{cls.__name__}] tool-calling detection: could not load a "
            f"tokenizer for {local_path}; treating as not tool-capable",
            exc_info=True,
        )
        return False
    if tokenizer is None:
        return False
    return tokenizer_declares_tools(tokenizer)

@classmethod
def compute_wire_tools(cls, local_path: Union[str, Path]) -> Optional[bool]:
    """Verified tool-call WIRE capability of the model at ``local_path`` (#298).

    ``compute_supports_tools`` says the chat template DECLARES tools; this
    says the engine's local server actually parses the model's tool-call
    output into a structured call the agent executes — a per-model wire
    property (#273 matrix): a template matching no parser streams the call
    as raw text (swallowed answer, #295) or leaks JSON to the user.

    Engine-specific: MLX runs mlx-vlm's own parser inference on the chat
    template; llama.cpp engines trust the ``--jinja`` generic fallback for
    any usable template. The base default is ``None`` (unverified) so an
    engine without an implementation never routes a KB turn agentic.
    """
    return None

@classmethod
def consume_brake_notice(cls) -> Optional[str]:
    """The brake's pending notice for the user, consumed. None here."""
    return None

@classmethod
def effective_context_tokens(cls) -> Optional[int]:
    """The ALLOCATED WINDOW: the context window of the CURRENTLY LOADED
    child, in tokens, or ``None`` when no child is up or the window could
    not be read.

    Read from the live engine handle (``_model``): llama.cpp engines stamp
    it from the server's ``/props`` after boot (the CPU engine's ``-c``,
    sized from the RAM budget at spawn and reduced by its memory brake; on
    CUDA, llama-server's own fit, which may shrink it below the model's
    trained window when memory demands); MLX stamps the trained window it passed at spawn as
    the preflight bound, then lowers it to the memory bound once the
    child's base footprint is measured, when memory binds
    (``memory_bound_tokens``). Never persisted anywhere -- the same model
    gets a different window on a different machine or memory state, so the
    only truthful source is the running child."""
    model = cls._model
    if isinstance(model, dict):
        window = model.get("context_tokens")
        if isinstance(window, int) and not isinstance(window, bool) and window > 0:
            return window
    return None

@classmethod
def end_memory_window(
    cls, token: Any, *, abandoned: bool = False, compaction_threshold: Optional[int] = None
) -> Any:
    """Close the window ``token`` opened and record what the turn used,
    with the compaction threshold of the conversation that ran it
    (``None``: an arena turn). No-op here."""
    return None

@classmethod
@asynccontextmanager
async def generation_guard(cls):
    """Serialize a full generation and suppress idle cleanup for its duration.

    The agent layer wraps model resolution + the entire token stream in this
    guard so that:
      - concurrent requests for different models can't thrash the
        single-model engine subprocess (they serialize on one asyncio lock);
      - the idle-cleanup tick (``_cleanup_tick``) shares this SAME lock, so
        it cannot run — let alone reap the model — while a generation is in
        flight. It waits until the guard exits, by which point the idle
        clock has been refreshed, so no mid-generation teardown is possible.

    This is the engine-level home of the invariant that ``generate_stream``
    used to carry; it lives here (not in the agent layer) to keep subprocess
    lifecycle and concurrency inside the engine encapsulation.

    The lock is taken through ``_acquire_generation_lock``, which first
    drains a prefix-cache reset an earlier holder left running.
    """
    lock = cls._generation_lock_for_running_loop()
    # Contention diagnostic: another generation (or the cleanup tick)
    # holds the lock right now — measure how long this request waits.
    wait_start_s = time.perf_counter()
    contended = lock.locked()
    async with cls._acquire_generation_lock():
        if contended:
            wait_ms = (time.perf_counter() - wait_start_s) * 1000
            logger.debug(
                f"[{cls.__name__}] generation_guard waited {wait_ms:.0f}ms "
                f"for the generation lock (model={cls._model_id})"
            )
        logger.info(f"[{cls.__name__}] generation_guard acquired (model={cls._model_id})")
        held_start_s = time.perf_counter()
        try:
            yield
        finally:
            # Restart the idle clock from the end of the generation.
            cls._last_used = datetime.now()
            held_ms = (time.perf_counter() - held_start_s) * 1000
            logger.info(
                f"[{cls.__name__}] generation_guard released "
                f"(model={cls._model_id}, held {held_ms:.0f}ms)"
            )

@classmethod
def get_engine(cls) -> Type["BaseEngine"]:
    """Select and return appropriate engine class for current platform.

    Automatically detects OS and hardware to choose the optimal backend:
    - macOS ARM (M1/M2/M3): MLX_Engine
    - macOS Intel: CPU_Engine
    - Linux/Windows with CUDA: CUDA_Engine
    - Linux/Windows without CUDA: CPU_Engine

    Returns:
        The engine class (not an instance) appropriate for this system.

    Raises:
        EngineException: If no suitable engine can be determined.

    Examples:
        from src.engines.base_engine import BaseEngine
        engine_class = BaseEngine.get_engine()
        print(f"Using: {engine_class}")  # e.g., "LLM Engine: MLX_Engine"

    """
    from src.engines.mlx_engine import MLX_Engine
    from src.engines.cpu_engine import CPU_Engine
    from src.engines.cuda_engine import CUDA_Engine

    system = platform.system().lower()  # mac, linux, windows
    machine = platform.machine().lower()  # arm64, x86_64, etc.
    llm_engine = None

    try:
        # Testing override: ERUDI_FORCE_CPU=1 bypasses GPU detection entirely.
        # Use this to validate CPU fallback on a GPU machine without disabling hardware.
        if os.environ.get("ERUDI_FORCE_CPU"):
            logger.info("[ERUDI_FORCE_CPU] Forcing CPU_Engine - GPU detection skipped.")
            return CPU_Engine

        if system == "darwin":  # MacOS
            if "arm" in machine:
                llm_engine = MLX_Engine
            elif "x86" in machine:
                llm_engine = CPU_Engine
        elif system in ("linux", "windows"):
            cuda_present = False
            try:
                import pynvml as nv

                nv.nvmlInit()
                cuda_present = nv.nvmlDeviceGetCount() > 0
            except ImportError:
                # A CPU build ships without pynvml: expected, not a defect.
                logger.info("pynvml is not installed; using CPU_Engine.")
            except Exception as e:
                if type(e).__name__ in _NO_NVIDIA_DRIVER_ERRORS:
                    # No NVIDIA driver on this machine: the ordinary way a
                    # CPU-only PC boots, not a fallback worth a warning.
                    logger.info(f"No NVIDIA driver ({e}); using CPU_Engine.")
                else:
                    # A driver that is present but broken. The user lands
                    # on the processor without being told why unless this
                    # record says so.
                    logger.warning(
                        f"NVIDIA detection failed ({type(e).__name__}: {e}); "
                        f"using CPU_Engine.",
                        exc_info=True,
                    )
            if cuda_present:
                llm_engine = CUDA_Engine
            else:
                llm_engine = CPU_Engine
                logger.info(f"System: {system} and CUDA not available.")
        logger.info(f"Engine chosen: {llm_engine}")
        if llm_engine is None:
            raise RuntimeError(f"no engine matches system={system!r} machine={machine!r}")
        return llm_engine
    except Exception as e:
        raise EngineException(
            message="Error selecting the LLM Engine.",
            trace=f"{type(e).__name__}: {e}",
        )

@classmethod
@abstractmethod
def get_flat_hardware_data(cls) -> Dict[str, Any]:
    """Get hardware data in flat format compatible with HardwareProfile entity.

    Returns hardware specifications as a flat dictionary ready for direct
    insertion into the HardwareProfile database entity. Combines and flattens
    data from get_hardware_info() and get_performance_evaluation().

    Returns:
        Dict with keys matching HardwareProfile columns:
        {
            # Common fields (all backends)
            "backend_type": str,  # "mlx", "cuda", "cpu"
            "cpu_model": str,
            "total_memory_gb": float,
            "available_memory_gb": float,
            "disk_total_gb": float,
            "disk_available_gb": float,
            "global_inference_score": float,
            "global_inference_label": str,
            "cpu_score": float,
            "memory_score": float,
            "gpu_score": float,
            "system_platform": str,
            "architecture": Optional[str],
            "estimated_tflops": Optional[float],
            "memory_bandwidth_gbs": Optional[float],
            "cpu_performance_units": Optional[float],
            "performance_breakdown": dict,

            # MLX-specific fields
            "mlx_chip_model": Optional[str],
            "mlx_gpu_cores": Optional[int],
            "mps_available": Optional[bool],
            "neural_engine_tops": Optional[float],
            "unified_memory": Optional[bool],
            "gpu_name": Optional[str],

            # CUDA-specific fields
            "cuda_cores": Optional[int],
            "cuda_version": Optional[str],
            "compute_capability": Optional[str],
            "vram_total_gb": Optional[float],
            "vram_available_gb": Optional[float],
        }

    Raises:
        EngineException: If hardware data collection fails.

    Note:
        This method provides a single point of access for hardware data
        in the format expected by the database layer, eliminating the need
        for manual flattening in service layers.

    Examples:
        >>> engine = BaseEngine.get_engine()
        >>> data = engine.get_flat_hardware_data()
        >>> profile = HardwareProfile(**data)  # Direct instantiation
        >>> db.add(profile)

    """
    pass

@classmethod
@abstractmethod
def get_hardware_info(cls) -> Dict[str, Any]:
    """Get comprehensive hardware information for this backend.

    Returns detailed hardware specifications including CPU, GPU/accelerator,
    memory, and platform-specific details. Implementation varies by backend:
    - MLX: Apple Silicon chip model, unified memory, MPS availability
    - CUDA: NVIDIA GPU model, CUDA cores, separate VRAM
    - CPU: Processor model, core count, system RAM

    Returns:
        Dict containing hardware specifications with the following structure:
        {
            "system": {
                "platform": str,  # "Darwin", "Linux", "Windows"
                "platform_version": Optional[str],
                "machine": str,  # "arm64", "x86_64", etc.
                "processor": str
            },
            "cpu": {
                "model": str,
                "architecture": str,
                "total_cores": int,
                "logical_cores": int,
                "is_apple_silicon": bool,  # MLX only
                "performance_cores": Optional[int],  # MLX only
                "efficiency_cores": Optional[int],  # MLX only
            },
            "memory": {
                "total_memory_gb": float,
                "available_memory_gb": float,
                "memory_pressure": float,  # 0.0-1.0
                "memory_type": str,  # "unified" (MLX) or "system" (CPU/CUDA)
            },
            "gpu": {
                "gpu_name": str,
                "gpu_cores": Optional[int],  # MLX: GPU cores, CUDA: CUDA cores
                "memory_bandwidth_gbs": Optional[float],
                "vram_total_gb": Optional[float],  # CUDA only
                "vram_available_gb": Optional[float],  # CUDA only
                "compute_capability": Optional[str],  # CUDA only
                "cuda_version": Optional[str],  # CUDA only
                "mps_supported": Optional[bool],  # MLX only
                "unified_memory": bool,  # True for MLX, False for CUDA/CPU
            },
            "accelerator": {  # MLX only
                "neural_engine_tops": Optional[float],
                "architecture": Optional[str],  # "3nm", "5nm", etc.
            },
            "storage": {
                "total_gb": float,
                "available_gb": float,
                "usage_percentage": float
            },
            "backend_type": str,  # "mlx", "cuda", or "cpu"
            "timestamp": float
        }

    Raises:
        EngineException: If hardware detection fails critically.

    Note:
        Implementation should handle errors gracefully and return fallback
        values rather than raising exceptions for non-critical failures.

    Examples:
        >>> engine = BaseEngine.get_engine()
        >>> hw_info = engine.get_hardware_info()
        >>> print(f"GPU: {hw_info['gpu']['gpu_name']}")
        >>> print(f"Total Memory: {hw_info['memory']['total_memory_gb']} GB")

    """
    pass

@classmethod
@abstractmethod
def get_model_and_tokenizer(
    cls, llm_id: str, llm_local_path: Union[str, Path], *args
) -> Tuple[Any, Any]:
    """Load or retrieve cached model and tokenizer for inference.

    Implements singleton pattern: returns cached model if already loaded
    for the given llm_id, otherwise loads from disk. Thread-safe.

    Args:
        llm_id: Unique identifier for the model (used for caching).
        llm_local_path: Path to the model directory on disk.
        *args: Engine-specific loading arguments (e.g., device, dtype).

    Returns:
        Tuple of (model, tokenizer) ready for inference.

    Raises:
        EngineException: If model loading fails.
        FileNotFoundError: If model path doesn't exist.

    Examples:
        from src.engines.base_engine import BaseEngine
        engine = BaseEngine.get_engine()
        model, tokenizer = engine.get_model_and_tokenizer(
            llm_id="1",
            llm_local_path="backend/data/models/llama-7b"
        )

    """
    pass

@classmethod
@abstractmethod
def get_performance_evaluation(cls) -> Dict[str, Any]:
    """Calculate comprehensive performance metrics and scores for this backend.

    Evaluates hardware capabilities and returns performance scores for
    inference workloads. Scoring methodology varies by backend but results
    are normalized to 0-100 scale for cross-platform comparison.

    Scoring Components:
        - **Inference Score**: Optimized for generation speed and latency.
          Weights: GPU/accelerator compute (35-60%), memory bandwidth (20-30%),
          memory capacity (10-30%), CPU (5-10%).

    Returns:
        Dict containing performance metrics and scores:
        {
            # Hardware identification
            "backend_type": str,  # "mlx", "cuda", "cpu"
            "accelerator_name": str,  # GPU/chip model
            "cpu_model": str,

            # Memory metrics
            "total_memory_gb": float,
            "available_memory_gb": float,
            "memory_bandwidth_gbs": Optional[float],

            # Storage metrics
            "disk_total_gb": float,
            "disk_available_gb": float,

            # Compute metrics
            "estimated_tflops": Optional[float],  # GPU compute power
            "compute_units": Optional[int],  # GPU cores or CUDA cores
            "cpu_performance_units": float,

            # Backend-specific metrics
            "neural_engine_tops": Optional[float],  # MLX only
            "cuda_version": Optional[str],  # CUDA only
            "compute_capability": Optional[str],  # CUDA only
            "architecture": Optional[str],  # MLX: "3nm", CUDA: "Ampere"

            # Performance scores (0-100)
            "global_inference_score": float,
            "global_inference_label": str,  # "Very Good", "Good", "Medium", "Poor"
            "gpu_score": float,
            "cpu_score": float,
            "memory_score": float,

            # Technical details
            "unified_memory": bool,
            "accelerator_available": bool,
            "system_platform": str,

            # Performance breakdown for debugging
            "performance_breakdown": {
                "compute_score": float,
                "memory_bandwidth_score": float,
                "memory_capacity_score": float,
                "cpu_performance_score": float,
                # Backend-specific breakdown
            }
        }

    Raises:
        EngineException: If evaluation fails critically.

    Note:
        Scores are platform-specific estimates based on hardware specs.
        Should call warm_up_accelerator() before evaluation for accuracy.
        Returns fallback scores if evaluation fails rather than raising.

    Examples:
        >>> engine = BaseEngine.get_engine()
        >>> engine.warm_up_accelerator(1.5)
        >>> eval_result = engine.get_performance_evaluation()
        >>> print(f"Inference: {eval_result['global_inference_score']}/100")

    """
    pass

@classmethod
def is_runnable(cls, link: str) -> bool:
    """Whether a catalog model (already engine-format by construction, since it
    came from a filter=FORMAT_TAG search) can actually run. The only exclusion
    is KNOWN_BROKEN — quants that load-crash on this engine."""
    return link not in cls.KNOWN_BROKEN

@classmethod
def max_context_tokens(cls) -> Optional[int]:
    """The DECLARED CEILING: the configured context window imposed on every
    spawn, or ``None`` when none is configured (the default everywhere).

    llama.cpp engines override this to return ``ERUDI_CTX`` when the user
    pinned one (then passed as ``-c``); MLX never imposes one. This is a
    configuration fact, not a runtime one -- for the window the loaded
    child actually runs with, read ``effective_context_tokens()``. The
    per-model sampling resolver (#388) caps ``max_tokens`` on this value
    when it exists."""
    return None

@classmethod
def memory_bound_footprint_bytes(cls) -> Optional[int]:
    """What a memory-bound child holds whatever the conversation does
    (the CPU engine's whole allocation at load: KV, compute, weights), the
    number the ``bound`` memory warning quotes when no per-token
    accounting exists; ``None`` here and on MLX, whose footprint grows
    with the conversation (``MemoryBudget.footprint_bytes``)."""
    return None

@classmethod
def memory_bound_tokens(cls) -> Optional[int]:
    """The window the machine's MEMORY imposed on the loaded child, or
    ``None`` when memory is not what bounds it.

    Set by an engine whose child's window memory limits: MLX, which lowers
    its child's window to the memory ceiling after load, the CPU engine,
    which sizes ``-c`` from the RAM budget at spawn (never for a
    sliding-window model, whose window the budget over-estimates), and
    the CUDA engine, which sizes it from the VRAM budget at spawn; it then
    equals the allocated window. ``None`` means the model's own
    window binds (or the bound could not be applied): no memory notice, no
    memory wording on an overflow and no memory reason on a compaction
    apply."""
    return None

@classmethod
def model_supports_vision(cls, local_path: Union[str, Path]) -> Optional[bool]:
    """Whether the model at ``local_path`` accepts image input (#133).

    Deterministic, no model load: llama.cpp engines look for an ``mmproj``
    projector, MLX reads ``config.json``. Returns ``None`` when the engine
    cannot decide; the runtime treats ``None`` as not vision-capable (#212):
    images are stripped unless the capability is an explicit ``True``,
    and the user is notified when the current turn carried an image.
    """
    return None

@classmethod
def note_call_start(cls, handle: Any, estimated_prompt_tokens: int) -> None:
    """A model call bound to ``handle`` is about to be sent, with the
    client's estimate of its size. No-op here."""
    return None

@classmethod
def note_call_usage(
    cls,
    handle: Any,
    input_tokens: int,
    output_tokens: int,
    cached_tokens: int = 0,
    has_images: bool = False,
) -> None:
    """One model call's server-reported usage, pushed by the chat client
    bound to ``handle``. No-op here."""
    return None

@classmethod
def note_stream_abandoned(cls, handle: Any) -> None:
    """A model stream against ``handle`` (the child it was built for) did
    not end normally. No-op here."""
    return None

@classmethod
def on_history_rewritten(cls) -> None:
    """The conversation history was rewritten (compaction): the cached
    prefix is garbage. No-op here."""
    return None

@classmethod
def start_cleanup_task(cls):
    """Start the automatic cleanup monitoring task.

    Creates an async task that periodically checks for idle models
    and frees memory. Should be called once during application startup.

    Note:
        Idempotent - calling multiple times has no effect if task already running.

    Examples:
        from src.engines.base_engine import BaseEngine
        engine = BaseEngine.get_engine()
        engine.start_cleanup_task()

    """
    if cls._cleanup_task is None:
        cls._cleanup_task = asyncio.create_task(cls._cleanup_monitor(), name="idle-cleanup")
        cls._cleanup_task.add_done_callback(_log_monitor_death)
        logger.info("Started cleanup monitor")

@classmethod
def start_pressure_task(cls) -> None:
    """Start the idle memory-pressure monitor. No-op here."""
    return None

@classmethod
def stop_cleanup_task(cls):
    """Stop the automatic cleanup monitoring task.

    Cancels the cleanup monitor task. Should be called during application
    shutdown to gracefully stop background tasks.

    Note:
        Idempotent - calling when no task is running has no effect.

    Examples:
        from src.engines.base_engine import BaseEngine
        engine = BaseEngine.get_engine()
        engine.stop_cleanup_task()

    """
    if cls._cleanup_task is not None:
        cls._cleanup_task.cancel()
        cls._cleanup_task = None
        logger.info("Stopped cleanup monitor")

@classmethod
async def stop_pressure_task(cls) -> None:
    """Stop the idle memory-pressure monitor and wait for it. No-op here."""
    return None

@classmethod
def take_idle_reload(cls, llm_id: Any) -> bool:
    """Whether this turn reloads ``llm_id`` after an idle-timeout reap.

    True once (the record is consumed) when the reap unloaded ``llm_id``
    and nothing is loaded. A record found while a model is loaded is stale
    (the title request of a new conversation reloaded it first) and is
    dropped. Identity is ``llm.id``, as in ``_should_not_reload_model``.
    """
    record = cls._idle_unload
    if not isinstance(record, dict):
        return False
    if cls._model is not None:
        cls._idle_unload = None
        return False
    if record.get("model_id") != llm_id:
        return False
    cls._idle_unload = None
    return True

@classmethod
@abstractmethod
def warm_up_accelerator(cls, duration_seconds: float = 1.0) -> bool:
    """Warm up the hardware accelerator (GPU/Neural Engine) for optimal performance.

    Runs compute-intensive operations to bring the accelerator to optimal
    performance state before benchmarking or inference. Implementation varies:
    - MLX: Matrix operations on MPS device
    - CUDA: CUDA kernel warm-up on GPU
    - CPU: CPU cache warm-up (minimal effect)

    Args:
        duration_seconds: How long to run warm-up operations (default: 1.0).

    Returns:
        bool: True if warm-up completed successfully, False otherwise.

    Raises:
        EngineException: If warm-up fails critically (rare, usually returns False).

    Note:
        This is particularly important for GPUs that dynamically adjust clocks.
        MLX/MPS benefits significantly from warm-up due to power management.
        CPU backend may implement minimal warm-up or skip entirely.

    Examples:
        >>> engine = BaseEngine.get_engine()
        >>> success = engine.warm_up_accelerator(1.5)
        >>> if success:
        ...     print("Accelerator ready for benchmarking")

    """
    pass

def __repr__(cls):
    """Return human-readable engine class name.

    Returns:
        String representation like "LLM Engine: MLX_Engine".

    """
    return f"LLM Engine: {cls.__name__}"

@classmethod
def brake_point(cls, point: str, *, llm_id: Any = None) -> BrakeOutcome:
    """Take the brake's sample at ``point`` and execute its decision.
    Never raises: a failure is ONE WARNING with the exception's class
    name, and the sample is lost."""
    try:
        return cls._llama_brake_point(point, llm_id)
    except Exception as exc:
        logger.warning(
            f"[{cls.__name__}] Memory brake sample failed ({type(exc).__name__}): "
            f"point={point}"
        )
        return BrakeOutcome()

@classmethod
def brake_reload_pending(cls, llm_id: Any) -> bool:
    """An unload's record for ``llm_id`` (the shared answer), or a brake
    respawn of its loaded child whose reload state is still owed."""
    return super().brake_reload_pending(llm_id) or cls._llama_respawn_pending(llm_id)

@classmethod
def get_flat_hardware_data(cls) -> Dict[str, Any]:
    """Get hardware data in flat format compatible with HardwareProfile entity.

    Returns hardware specifications as a flat dictionary ready for database
    insertion. For CPU backend, get_performance_evaluation() already returns
    data in the correct flat format.

    Returns:
        Flat dict with all fields matching HardwareProfile columns.

    Raises:
        HardwareException: If hardware data collection fails.
    """
    # get_performance_evaluation() already returns flat structure
    return cls.get_performance_evaluation()