def prompt_llm(
    system_prompt: str,
    user_prompt: str,
    formatting: FormattingType | None = None,
) -> str | dict[str, Any]:
    """
    Run a prompt on the tasks model.
    """
    from uni_server.app_state import get_uni  # noqa: PLC0415 - circular import

    uni = get_uni()

    messages = [
        {"role": "system", "content": textwrap.dedent(system_prompt).strip()},
        {"role": "user", "content": textwrap.dedent(user_prompt).strip()},
    ]

    chunks = list(
        uni.model_router.chat(
            role="tasks",
            messages=messages,
            formatting=formatting,
        )
    )

    if not chunks:
        raise RuntimeError("No response from LLM provider")

    output = "".join(chunk["message"]["content"] for chunk in chunks).strip()

    if formatting:
        return parse_llm_generated_json(output)
    return output

def parse_datetime(text: str, prefer_future: bool = True) -> datetime:
    """
    Parse natural language datetime strings (e.g., "tomorrow at 3pm", "next Monday").
    """
    settings: dict[str, Any] = {
        "PREFER_DAY_OF_MONTH": "first",  # "December" -> Dec 1st, not Dec 16th
    }
    if prefer_future:
        settings["PREFER_DATES_FROM"] = "future"

    result = dateparser.parse(text, settings=cast(Any, settings))
    if result is None:
        raise ValueError(f"Could not parse '{text}' as a datetime")
    return result

def parse_duration(text: str) -> int:
    """
    Parse natural language duration strings (e.g., "2 hours", "1h 30m", "45s") to seconds.
    """
    if text.strip().startswith("-"):
        raise ValueError(f"Duration '{text}' must be positive")

    reference_time = datetime(2025, 1, 1, 12, 0, 0)
    future_time = dateparser.parse(
        f"in {text}",
        settings={
            "PREFER_DATES_FROM": "future",
            "PREFER_DAY_OF_MONTH": "first",
            "RELATIVE_BASE": reference_time,
        },
    )
    if future_time is None:
        raise ValueError(f"Could not parse '{text}' as a duration")

    duration_seconds = int((future_time - reference_time).total_seconds())
    if duration_seconds < 0:
        raise ValueError(f"Duration '{text}' must be positive")

    return duration_seconds

def serve_file(base_dir: Path, relative_path: str) -> FileResponse:
    """
    Serve a file, ensuring the resolved path stays within base_dir.

    Replaces bare FileResponse(base_dir / user_input) which is vulnerable
    to path traversal attacks. Starlette's FileResponse does not validate
    that the resolved path stays within the base directory.
    """
    resolved = (base_dir / relative_path).resolve()
    if not resolved.is_relative_to(base_dir.resolve()):
        raise HTTPException(status_code=404)
    if not resolved.is_file():
        raise HTTPException(status_code=404)
    return FileResponse(resolved)

def html_to_markdown(
    html: str,
    preserve_links: bool = True,
    truncate_after: int | None = None,
    body_width: int = 0,
) -> str:
    """
    Convert HTML to clean Markdown for LLM consumption.

    Strips images and excessive formatting while preserving semantic structure.
    """
    if not html or not html.strip():
        return ""

    h = html2text.HTML2Text()

    h.ignore_links = not preserve_links
    h.ignore_images = True
    h.ignore_emphasis = False
    h.body_width = body_width
    h.unicode_snob = True
    h.escape_snob = True

    markdown_text = h.handle(html)
    markdown_text = _clean_markdown_text(markdown_text)

    if truncate_after and len(markdown_text) > truncate_after:
        markdown_text = markdown_text[:truncate_after].rstrip() + "...[truncated]"

    return markdown_text

def safe_filename(filename: str) -> str:
    r"""
    Return a secure version of a filename for safe filesystem storage.

    The result can safely be passed to :func:`os.path.join`. The filename
    returned is ASCII-only for maximum portability. On Windows the function
    also avoids special device file names.

    >>> safe_filename("My cool movie.mov")
    'My_cool_movie.mov'
    >>> safe_filename("../../../etc/passwd")
    'etc_passwd'
    >>> safe_filename('i contain cool \xfcml\xe4uts.txt')
    'i_contain_cool_umlauts.txt'

    May return an empty string if the input contains only unsafe characters.
    """
    filename = unicodedata.normalize("NFKD", filename)
    filename = filename.encode("ascii", "ignore").decode("ascii")

    for sep in os.sep, os.path.altsep:
        if sep:
            filename = filename.replace(sep, " ")

    filename = _filename_ascii_strip_re.sub("", "_".join(filename.split())).strip("._")

    if (
        os.name == "nt"
        and filename
        and filename.split(".")[0].upper() in _windows_device_files
    ):
        filename = f"_{filename}"

    return filename

def is_cuda_available() -> bool:
    """
    Check if CUDA is available for PyTorch operations.

    Returns:
        True if CUDA is available and PyTorch was built with CUDA support, False otherwise.
    """
    import torch  # noqa: PLC0415 - heavy import, only needed here

    return torch.cuda.is_available()

def is_hf_authenticated() -> bool:
    """
    Check if Hugging Face authentication is active.

    Returns:
        True if authenticated with Hugging Face, False otherwise.
    """
    return _plugin_state.is_hf_authenticated()

def flag_for_restart() -> None:
    """
    Flag that a restart is needed to apply settings changes.

    Call this when the user changes a setting that requires a plugin reload to take
    effect. This activates the "Restart to apply changes" banner.
    """
    restart_state.flag_for_restart()

def get_config() -> AppConfig:
    """
    Get the central application configuration from `/user/settings.toml`.

    Use `config.plugin` to access settings specific to the current plugin.
    """
    return _plugin_state.get_config()

def llm_chat(
    role: str,
    messages: list[dict[str, Any]],
    **overrides: Any,
) -> Iterator[dict[str, Any]]:
    """
    Stream chat completions from the model assigned to ``role``.
    """
    router = _get_model_router()
    return router.chat(role=role, messages=messages, **overrides)

def llm_embeddings(text: str) -> list[float]:
    """
    Generate embeddings using the configured embeddings model.
    """
    router = _get_model_router()
    return router.embeddings(text)

def llm_embeddings_model() -> str | None:
    """
    Return the configured embeddings model name, or None if not configured.
    """
    router = _get_model_router()
    return router.embeddings_model()

def cache(
    ttl: str | None = None,
    max_size: int | None = None,
) -> Callable[[Callable[P, R]], CachedFunction[P, R]]:
    """
    Decorator to cache function results with optional TTL and size limit.

    This decorator can be used to effectively rate limit expensive operations
    by caching their results for a specified duration.

    ```python
    @uni.cache(ttl="5m")
    def fetch_weather_data():
        return {"temperature": 22, "conditions": "Sunny"}

    @uni.cache(ttl="1h", max_size=100)
    def get_user_profile(user_id: str):
        return {"id": user_id, "name": "..."}
    ```

    Args:
        ttl: Time-to-live for cached results (e.g., "5m", "1h", None = cache forever)
        max_size: Maximum number of cached entries (None = unlimited)
    """

    def decorator(func: Callable[P, R]) -> CachedFunction[P, R]:
        cache_data: dict[tuple[Any, ...], tuple[R, float]] = {}
        ttl_seconds = parse_duration(ttl) if ttl is not None else None

        @wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            try:
                key: tuple[Any, ...] = (args, tuple(sorted(kwargs.items())))
                hash(key)
            except TypeError:
                return func(*args, **kwargs)

            if key in cache_data:
                result, timestamp = cache_data[key]
                if ttl_seconds is None or get_time() - timestamp < ttl_seconds:
                    return result
                del cache_data[key]

            if max_size is not None and len(cache_data) >= max_size:
                oldest_key = next(iter(cache_data))
                del cache_data[oldest_key]

            result = func(*args, **kwargs)
            cache_data[key] = (result, get_time())
            return result

        def clear_cache() -> None:
            cache_data.clear()

        wrapper_with_attr = cast(Any, wrapper)
        wrapper_with_attr.clear_cache = clear_cache
        return cast(CachedFunction[P, R], wrapper_with_attr)

    return decorator

def socket(handler_func: Callable[..., None]) -> Callable[..., None]:
    """
    Decorator to register Socket.IO handlers for this plugin.

    ```python
    @uni.socket
    def register_socket_handlers(socket):
        @socket.on("game_move")
        def handle_move(data):
            # uni.current_device.id automatically available
            socket.emit("move_result", {"success": True})

        @socket.on_connect
        def handle_connect():
            socket.emit("welcome", {"message": "Connected!"})
    ```
    """
    get_uni().plugins.register_socket_handlers(handler_func)
    return handler_func

def tool[T: Callable[..., Any]](
    func: T | None = None,
    *,
    safe: bool | Callable[..., bool] = False,
) -> T | Callable[[T], T]:
    """
    Decorator to register a function as a tool for the LLM to use.

    Args:
        func: The tool function (when used without parentheses)
        safe: Whether the tool is safe to execute without approval.
              Can be True (always safe), False (always needs approval),
              or a callable that validates safety based on arguments.

    ```python
    # Default: requires approval
    @uni.tool
    def send_email(to: str, subject: str, body: str): ...

    # Always safe
    @uni.tool(safe=True)
    def calculate(expression: str): ...

    # Conditionally safe based on arguments
    def is_safe_url(url: str) -> bool:
        return url in get_recent_search_urls()

    @uni.tool(safe=is_safe_url)
    def fetch_webpage(url: str): ...
    ```
    """

    def register(f: T) -> T:
        get_uni().plugins.register_tool(f, safe=safe)
        return f

    if func is None:
        return register
    else:
        return register(func)

def register_tool(
    schema: dict[str, Any],
    handler: Callable[..., Any],
    /,
    safe: bool | Callable[..., bool] = False,
) -> None:
    """
    Register a tool with a custom JSON schema.

    Args:
        schema: Dictionary describing the tool JSON schema
        handler: Function to handle tool calls
        safe: Whether the tool is safe to execute without approval
    """
    get_uni().plugins.register_custom_tool(
        ToolDescriptor(schema=schema, handler=handler),
        safe=safe,
    )

def audio_effect(
    effect: type[AudioEffect] | AudioEffect,
) -> type[AudioEffect] | AudioEffect:
    """
    Decorator to register an audio effect for TTS output.

    ```python
    @uni.audio_effect
    class MyEffect(uni.AudioEffect):
        ...
    ```
    """
    effect_instance = _ensure_instance(effect, AudioEffect)  # type: ignore[type-abstract]
    get_uni().plugins.register_audio_effect(effect_instance)
    return effect

def system_prompt_modifier(
    modifier: SystemPromptModifierFunction | None = None,
    *,
    priority: int = 100,
) -> (
    SystemPromptModifierFunction
    | Callable[[SystemPromptModifierFunction], SystemPromptModifierFunction]
):
    """
    Decorator to register a function that modifies the system prompt.

    ```python
    @uni.system_prompt_modifier
    def my_modifier(event: uni.SystemPromptEvent) -> str:
        return event.prompt + "\\n\\nAdditional instructions here."
    ```
    """

    def register(
        func: SystemPromptModifierFunction,
    ) -> SystemPromptModifierFunction:
        get_uni().plugins.register_system_prompt_modifier(func, priority)
        return func

    if modifier is None:
        return register
    else:
        return register(modifier)

def user_prompt_modifier(
    modifier: UserPromptModifierFunction | None = None,
    *,
    priority: int = 100,
) -> (
    UserPromptModifierFunction
    | Callable[[UserPromptModifierFunction], UserPromptModifierFunction]
):
    """
    Decorator to register a function that modifies individual user prompts.

    The return value replaces the entire prompt, so include the original:

    ```python
    @uni.user_prompt_modifier
    def add_context(event: UserPromptEvent) -> str | None:
        if "weather" in event.prompt:
            return event.prompt + "\nRemember to provide current weather info."
        return None
    ```

    Args:
        modifier: Function that takes a UserPromptEvent and returns modified prompt or None
        priority: Execution priority (lower values run first, default: 100)
    """

    def register(
        func: UserPromptModifierFunction,
    ) -> UserPromptModifierFunction:
        get_uni().plugins.register_user_prompt_modifier(func, priority)
        return func

    if modifier is None:
        return register
    else:
        return register(modifier)

def message_context(
    func: MessageContextFunction | None = None,
    *,
    preload: Callable[[], Any] | None = None,
) -> MessageContextFunction | Callable[[MessageContextFunction], MessageContextFunction]:
    """
    Decorator to inject ephemeral context for a specific message.

    ```python
    @uni.message_context
    def inject_docs(event: MessageEvent) -> str | None:
        if needs_docs(event.message):
            return "Relevant documentation: ..."
        return None

    # With preloading
    @uni.message_context(preload=lambda: fetch_weather())
    def inject_weather(event: MessageEvent, weather_data: dict) -> dict | None:
        if needs_weather(event.message):
            return weather_data  # Can return dict directly
        return None
    ```

    Args:
        func: Function that takes MessageEvent (and preloaded data if applicable)
        preload: Optional function to preload data while user is speaking
    """

    def register(
        context_func: MessageContextFunction,
    ) -> MessageContextFunction:
        get_uni().plugins.register_message_context(context_func, preload)
        return context_func

    if func is None:
        return register
    else:
        return register(func)

def on_nightly_reset(
    func: Callable[[NightlyResetEvent], None],
) -> Callable[[NightlyResetEvent], None]:
    """
    Decorator to register a function to be called during nightly reset.

    ```python
    @uni.on_nightly_reset
    def consolidate_data(event: NightlyResetEvent) -> None:
        # Process the day's conversation history
        process_history(event.conversation_history)
    ```
    """
    get_uni().plugins.events.register_nightly_reset_handler(func)
    return func

def tts(
    engine: type[TtsEngine] | TtsEngine,
) -> type[TtsEngine] | TtsEngine:
    """
    Decorator to override the default TTS engine.

    ```python
    @uni.tts
    class MyTtsEngine(uni.TtsEngine):
        ...
    ```
    """
    tts_instance = _ensure_instance(engine, TtsEngine)  # type: ignore[type-abstract]
    get_uni().plugins.register_tts_engine(tts_instance)
    return engine

def wake_word(
    engine: type[WakeWordEngine] | WakeWordEngine,
) -> type[WakeWordEngine] | WakeWordEngine:
    """
    Decorator to register a wake word detection engine.

    ```python
    @uni.wake_word
    class MyWakeWordEngine(uni.WakeWordEngine):
        ...
    ```
    """
    engine_instance = _ensure_instance(engine, WakeWordEngine)  # type: ignore[type-abstract]
    get_uni().plugins.register_wake_word_engine(engine_instance)
    return engine

def avatar(module: str) -> Callable[[AvatarConfigFunction], AvatarConfigFunction]:
    """
    Decorator to register an avatar provider.

    ```python
    @uni.avatar(module="~/avatar.js")
    def create_avatar() -> dict[str, Any]:
        # Return a configuration object for the frontend
        return {
            "expressions": ["normal", "happy", "listening", "sleeping"]
        }
    ```

    If an `expressions` key is present, those expressions will be triggered automatically
    during the conversation based on sentiment analysis.

    Refer to the plugins documentation for the JS module interface.

    Args:
        module: Path to the JavaScript module. `"~/"` = plugin static root
    """

    def register(func: AvatarConfigFunction) -> AvatarConfigFunction:
        if module.startswith("~/"):
            plugin_name = find_current_plugin_name()
            module_path = f"/plugins/{plugin_name}/static/{module[2:]}"
            static_path = f"/plugins/{plugin_name}/static"
        else:
            module_path = module
            static_path = ""

        get_uni().plugins.register_avatar(func, module_path, static_path)
        return func

    return register

def message_channel(
    cls: type[MessageChannel] | MessageChannel,
) -> type[MessageChannel] | MessageChannel:
    """
    Decorator to register a message channel for async communication.

    ```python
    @uni.message_channel
    class DiscordChannel(uni.MessageChannel):
        @property
        def online(self) -> bool:
            return self.client.is_ready()

        def send(self, message: str) -> None:
            self.dm_channel.send(message)
    ```
    """
    instance = _ensure_instance(cls, MessageChannel)  # type: ignore[type-abstract]
    get_uni().plugins.register_message_channel(instance)
    return cls