"""llama.cpp Provider for Frigate AI."""

import base64
import io
import json
import logging
from collections.abc import AsyncGenerator
from typing import Any, cast

import httpx
import numpy as np
import requests
from PIL import Image

from frigate.config import GenAIProviderEnum
from frigate.genai import GenAIClient, register_genai_provider
from frigate.genai.utils import parse_tool_calls_from_message

logger = logging.getLogger(__name__)


def _stats_from_llama_cpp_chunk(data: dict[str, Any]) -> dict[str, Any] | None:
    """Build a stats dict from a llama.cpp streaming chunk.

    Final-chunk `usage` carries authoritative token counts. Per-chunk
    `timings` (enabled via timings_per_token) carries the running token
    counts (prompt_n, predicted_n) and generation rate, so live updates
    work mid-stream.
    """
    usage = data.get("usage") or {}
    timings = data.get("timings") or {}
    prompt_tokens = usage.get("prompt_tokens")
    completion_tokens = usage.get("completion_tokens")
    predicted_ms = timings.get("predicted_ms")
    tps = timings.get("predicted_per_second")
    stats: dict[str, Any] = {}

    if not isinstance(prompt_tokens, int):
        prompt_n = timings.get("prompt_n")

        if isinstance(prompt_n, int):
            prompt_tokens = prompt_n

    if not isinstance(completion_tokens, int):
        predicted_n = timings.get("predicted_n")

        if isinstance(predicted_n, int):
            completion_tokens = predicted_n

    if not isinstance(prompt_tokens, int) and not isinstance(completion_tokens, int):
        return None

    if isinstance(prompt_tokens, int):
        stats["prompt_tokens"] = prompt_tokens

    if isinstance(completion_tokens, int):
        stats["completion_tokens"] = completion_tokens

    if isinstance(predicted_ms, (int, float)) and predicted_ms > 0:
        stats["completion_duration_ms"] = float(predicted_ms)

    if isinstance(tps, (int, float)) and tps > 0:
        stats["tokens_per_second"] = float(tps)

    return stats or None


def _parse_launch_arg(args: list[str], flag: str) -> str | None:
    """Return the value following `flag` in a positional argv list, or None."""
    try:
        idx = args.index(flag)
    except ValueError:
        return None
    if idx + 1 >= len(args):
        return None
    return args[idx + 1]


def _to_jpeg(img_bytes: bytes) -> bytes | None:
    """Convert image bytes to JPEG. llama.cpp/STB does not support WebP."""
    try:
        img = Image.open(io.BytesIO(img_bytes))
        if img.mode != "RGB":
            img = img.convert("RGB")  # type: ignore[assignment]
        buf = io.BytesIO()
        img.save(buf, format="JPEG", quality=85)
        return buf.getvalue()
    except Exception as e:
        logger.warning("Failed to convert image to JPEG: %s", e)
        return None


@register_genai_provider(GenAIProviderEnum.llamacpp)
class LlamaCppClient(GenAIClient):
    """Generative AI client for Frigate using llama.cpp server."""

    provider: str | None  # base_url
    provider_options: dict[str, Any]
    _context_size: int | None
    _supports_vision: bool
    _supports_audio: bool
    _supports_tools: bool
    _supports_reasoning: bool
    _image_token_cache: dict[tuple[int, int], int]
    _text_baseline_tokens: int | None
    _media_marker: str

    @property
    def supports_embeddings(self) -> bool:
        """llama.cpp exposes an /embeddings endpoint for any loaded model."""
        return True

    def _auth_headers(self) -> dict | None:
        """Bearer auth header when an API key is configured, else None."""
        if self.genai_config.api_key:
            return {"Authorization": "Bearer " + self.genai_config.api_key}

        return None

    def _get(self, url: str, **kwargs: Any) -> requests.Response:
        """GET with the configured auth headers injected."""
        return requests.get(url, headers=self._auth_headers(), **kwargs)

    def _post(self, url: str, **kwargs: Any) -> requests.Response:
        """POST with the configured auth headers injected."""
        return requests.post(url, headers=self._auth_headers(), **kwargs)

    def _fetch_llama_props(self, base_url: str, model: str) -> dict[str, Any]:
        """Fetch /props from a llama.cpp server, with llama-swap fallback.

        Raises the underlying RequestException if both endpoints fail; callers
        decide how to surface the failure.
        """
        try:
            response = self._get(
                f"{base_url}/props",
                params={"model": model},
                timeout=10,
            )
            response.raise_for_status()
            return cast(dict[str, Any], response.json())
        except Exception:
            response = self._get(
                f"{base_url}/upstream/{model}/props",
                timeout=10,
            )
            response.raise_for_status()
            return cast(dict[str, Any], response.json())

    def _init_provider(self) -> str | None:
        """Initialize the client and query model metadata from the server."""
        self.provider_options = {
            **self.genai_config.provider_options,
        }
        self._context_size = None
        self._supports_vision = False
        self._supports_audio = False
        self._supports_tools = False
        self._supports_reasoning = False
        self._image_token_cache = {}
        self._text_baseline_tokens = None
        self._media_marker = "<__media__>"

        base_url = (
            self.genai_config.base_url.rstrip("/")
            if self.genai_config.base_url
            else None
        )

        if base_url is None:
            return None
        else:
            base_url = base_url.replace("/v1", "")  # Strip /v1 if included in base_url

        if not self.validate_model:
            # Probe path
            return base_url

        configured_model = self.genai_config.model
        info = self._get_model_info(base_url, configured_model)

        if info is None:
            return None

        self._context_size = info["context_size"]
        self._supports_vision = info["supports_vision"]
        self._supports_audio = info["supports_audio"]
        self._supports_tools = info["supports_tools"]
        self._supports_reasoning = info["supports_reasoning"]
        self._media_marker = info["media_marker"]

        logger.info(
            "llama.cpp model '%s' initialized — context: %s, vision: %s, audio: %s, tools: %s, reasoning: %s",
            configured_model,
            self.get_context_size(),
            self._supports_vision,
            self._supports_audio,
            self._supports_tools,
            self._supports_reasoning,
        )

        return base_url

    def _get_model_info(
        self, base_url: str, configured_model: str
    ) -> dict[str, Any] | None:
        """Resolve model metadata from /v1/models with /props fallback.

        Returns a dict of capability fields, or None if the server's model
        registry was reachable and reported the configured model as missing.
        A reachable-but-unparseable /v1/models is treated as soft-pass and
        falls through to /props, matching prior behavior.

        After ggml-org/llama.cpp#22952, /v1/models exposes per-model
        `architecture.input_modalities` (text/image/audio) — the primary
        source. When proxied through llama-swap, the same entry carries
        `status.args` (server launch argv) and, for the loaded model,
        `meta.n_ctx`. /props remains the only source for `media_marker`,
        which the server randomizes per startup unless LLAMA_MEDIA_MARKER
        is set.
        """
        info: dict[str, Any] = {
            "context_size": None,
            "supports_vision": False,
            "supports_audio": False,
            "supports_tools": False,
            "supports_reasoning": False,
            "media_marker": "<__media__>",
        }

        model_entry: dict[str, Any] | None = None
        try:
            response = self._get(f"{base_url}/v1/models", timeout=10)
            response.raise_for_status()
            models_data = response.json()

            for model in models_data.get("data", []):
                model_ids = {model.get("id")}
                for alias in model.get("aliases", []):
                    model_ids.add(alias)
                if configured_model in model_ids:
                    model_entry = model
                    break

            if model_entry is None:
                available = []
                for m in models_data.get("data", []):
                    available.append(m.get("id", "unknown"))
                    for alias in m.get("aliases", []):
                        available.append(alias)
                logger.error(
                    "Model '%s' not found on llama.cpp server. Available models: %s",
                    configured_model,
                    available,
                )
                return None
        except Exception as e:
            logger.warning(
                "Failed to query llama.cpp /v1/models endpoint: %s. "
                "Model validation skipped.",
                e,
            )

        if model_entry is not None:
            architecture = model_entry.get("architecture") or {}
            input_modalities = architecture.get("input_modalities") or []

            if isinstance(input_modalities, list):
                info["supports_vision"] = "image" in input_modalities
                info["supports_audio"] = "audio" in input_modalities

            status = model_entry.get("status") or {}
            launch_args = status.get("args") if isinstance(status, dict) else None
            if not isinstance(launch_args, list):
                launch_args = []

            meta = model_entry.get("meta") if isinstance(model_entry, dict) else None
            n_ctx = meta.get("n_ctx") if isinstance(meta, dict) else None

            if not n_ctx:
                n_ctx = _parse_launch_arg(launch_args, "--ctx-size")

            if n_ctx:
                try:
                    info["context_size"] = int(n_ctx)
                except (TypeError, ValueError):
                    pass

            # Tool calling on llama-server requires --jinja.
            if "--jinja" in launch_args:
                info["supports_tools"] = True

        try:
            props = self._fetch_llama_props(base_url, configured_model)

            if info["context_size"] is None:
                default_settings = props.get("default_generation_settings", {})
                n_ctx = default_settings.get("n_ctx")
                if n_ctx:
                    info["context_size"] = int(n_ctx)

            if not (info["supports_vision"] or info["supports_audio"]):
                modalities = props.get("modalities", {})
                info["supports_vision"] = bool(modalities.get("vision", False))
                info["supports_audio"] = bool(modalities.get("audio", False))

            chat_caps = props.get("chat_template_caps") or {}

            if not info["supports_tools"]:
                info["supports_tools"] = bool(chat_caps.get("supports_tools", False))

            # llama.cpp does not advertise per-template reasoning support, so
            # detect it by looking for the `enable_thinking` toggle variable
            # in the Jinja chat template itself.
            chat_template = props.get("chat_template") or ""
            info["supports_reasoning"] = "enable_thinking" in chat_template

            media_marker = props.get("media_marker")
            if isinstance(media_marker, str) and media_marker:
                info["media_marker"] = media_marker
        except Exception as e:
            logger.warning(
                "Failed to query llama.cpp /props endpoint: %s. "
                "Image embeddings may fail if the server randomized its media marker.",
                e,
            )

        return info

    def _send(
        self,
        prompt: str,
        images: list[bytes],
        response_format: dict | None = None,
        enable_thinking: bool = False,
    ) -> str | None:
        """Submit a request to llama.cpp server."""
        if self.provider is None:
            logger.warning(
                "llama.cpp provider has not been initialized, a description will not be generated. Check your llama.cpp configuration."
            )
            return None

        try:
            content = [
                {
                    "type": "text",
                    "text": prompt,
                }
            ]
            for image in images:
                encoded_image = base64.b64encode(image).decode("utf-8")
                content.append(
                    {
                        "type": "image_url",
                        "image_url": {  # type: ignore[dict-item]
                            "url": f"data:image/jpeg;base64,{encoded_image}",
                        },
                    }
                )

            # Build request payload with llama.cpp native options
            payload: dict[str, Any] = {
                "model": self.genai_config.model,
                "messages": [
                    {
                        "role": "user",
                        "content": content,
                    },
                ],
                **self.provider_options,
            }

            if response_format:
                payload["response_format"] = response_format

            if self.supports_toggleable_thinking:
                payload["chat_template_kwargs"] = {"enable_thinking": enable_thinking}

            response = self._post(
                f"{self.provider}/v1/chat/completions",
                json=payload,
                timeout=self.timeout,
            )
            response.raise_for_status()
            result = response.json()

            if (
                result is not None
                and "choices" in result
                and len(result["choices"]) > 0
            ):
                choice = result["choices"][0]
                if "message" in choice and "content" in choice["message"]:
                    return str(choice["message"]["content"].strip())
            return None
        except Exception as e:
            logger.warning("llama.cpp returned an error: %s", str(e))
            return None

    @property
    def supports_vision(self) -> bool:
        """Whether the loaded model supports vision/image input."""
        return self._supports_vision

    @property
    def supports_audio(self) -> bool:
        """Whether the loaded model supports audio input."""
        return self._supports_audio

    @property
    def supports_tools(self) -> bool:
        """Whether the loaded model supports tool/function calling."""
        return self._supports_tools

    @property
    def supports_toggleable_thinking(self) -> bool:
        return self._supports_reasoning

    def list_models(self) -> list[str]:
        """Return available model IDs from the llama.cpp server."""
        base_url = self.provider or (
            self.genai_config.base_url.rstrip("/")
            if self.genai_config.base_url
            else None
        )
        if base_url is None:
            return []
        try:
            response = self._get(f"{base_url}/v1/models", timeout=10)
            response.raise_for_status()
            models = []
            for m in response.json().get("data", []):
                models.append(m.get("id", "unknown"))
                for alias in m.get("aliases", []):
                    models.append(alias)
            return sorted(models)
        except Exception as e:
            logger.warning("Failed to list llama.cpp models: %s", e)
            return []

    def get_context_size(self) -> int:
        """Get the context window size for llama.cpp.

        Resolution order:
        1. provider_options["context_size"] (user override)
        2. Value queried from llama.cpp server at init
        3. Default fallback of 4096
        """
        if "context_size" in self.provider_options:
            return int(self.provider_options["context_size"])
        if self._context_size is not None:
            return self._context_size
        return 4096

    def estimate_image_tokens(self, width: int, height: int) -> float:
        """Probe the llama.cpp server to learn the model's image-token cost at the
        requested dimensions.

        llama.cpp's image tokenization is a deterministic function of dimensions and
        the loaded mmproj, so the result is cached per (width, height) for the
        lifetime of the process. Falls back to the base pixel heuristic if the
        server is unreachable or the response is malformed.
        """
        if self.provider is None:
            return super().estimate_image_tokens(width, height)

        cached = self._image_token_cache.get((width, height))

        if cached is not None:
            return cached

        try:
            baseline = self._probe_baseline_tokens()
            with_image = self._probe_image_prompt_tokens(width, height)
            tokens = max(1, with_image - baseline)
        except Exception as e:
            logger.debug(
                "llama.cpp image-token probe failed for %dx%d (%s); using heuristic",
                width,
                height,
                e,
            )
            return super().estimate_image_tokens(width, height)

        self._image_token_cache[(width, height)] = tokens
        logger.debug(
            "llama.cpp model '%s' uses ~%d tokens for %dx%d images",
            self.genai_config.model,
            tokens,
            width,
            height,
        )
        return tokens

    def _probe_baseline_tokens(self) -> int:
        """Return prompt_tokens for a minimal text-only request. Cached after first call."""
        if self._text_baseline_tokens is not None:
            return self._text_baseline_tokens

        self._text_baseline_tokens = self._probe_prompt_tokens(
            [{"type": "text", "text": "."}]
        )
        return self._text_baseline_tokens

    def _probe_image_prompt_tokens(self, width: int, height: int) -> int:
        """Return prompt_tokens for a single synthetic image plus minimal text."""
        img = Image.new("RGB", (width, height), (128, 128, 128))
        buf = io.BytesIO()
        img.save(buf, format="JPEG", quality=60)
        encoded = base64.b64encode(buf.getvalue()).decode("utf-8")
        return self._probe_prompt_tokens(
            [
                {"type": "text", "text": "."},
                {
                    "type": "image_url",
                    "image_url": {"url": f"data:image/jpeg;base64,{encoded}"},
                },
            ]
        )

    def _probe_prompt_tokens(self, content: list[dict[str, Any]]) -> int:
        """POST a 1-token chat completion and return reported prompt_tokens.

        Uses a generous timeout to absorb a cold model load on the first probe
        when the server lazily loads models on demand (e.g. llama-swap).
        """
        payload = {
            "model": self.genai_config.model,
            "messages": [{"role": "user", "content": content}],
            "max_tokens": 1,
        }
        response = self._post(
            f"{self.provider}/v1/chat/completions",
            json=payload,
            timeout=60,
        )
        response.raise_for_status()
        return int(response.json()["usage"]["prompt_tokens"])

    def _build_payload(
        self,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]] | None,
        tool_choice: str | None,
        stream: bool = False,
        enable_thinking: bool | None = None,
    ) -> dict[str, Any]:
        """Build request payload for chat completions (sync or stream)."""
        openai_tool_choice = None
        if tool_choice:
            if tool_choice == "none":
                openai_tool_choice = "none"
            elif tool_choice == "auto":
                openai_tool_choice = "auto"
            elif tool_choice == "required":
                openai_tool_choice = "required"

        payload: dict[str, Any] = {
            "messages": messages,
            "model": self.genai_config.model,
        }

        if stream:
            payload["stream"] = True
            payload["stream_options"] = {"include_usage": True}
            payload["timings_per_token"] = True

        if tools:
            payload["tools"] = tools

            if openai_tool_choice is not None:
                payload["tool_choice"] = openai_tool_choice

        if enable_thinking is not None and self._supports_reasoning:
            payload["chat_template_kwargs"] = {"enable_thinking": enable_thinking}

        provider_opts = {
            k: v for k, v in self.provider_options.items() if k != "context_size"
        }
        payload.update(provider_opts)
        payload.update(self.genai_config.runtime_options)
        return payload

    def _message_from_choice(self, choice: dict[str, Any]) -> dict[str, Any]:
        """Parse OpenAI-style choice into {content, reasoning, tool_calls, finish_reason}.

        llama.cpp's `--reasoning-format` puts the trace in
        `message.reasoning_content` (preferred) or `message.thinking`; both
        keys are accepted so different builds work without configuration.
        """
        message = choice.get("message", {})
        content = message.get("content")
        content = content.strip() if content else None
        reasoning = message.get("reasoning_content") or message.get("thinking")
        reasoning = reasoning.strip() if reasoning else None
        tool_calls = parse_tool_calls_from_message(message)
        finish_reason = choice.get("finish_reason") or (
            "tool_calls" if tool_calls else "stop" if content else "error"
        )
        return {
            "content": content,
            "reasoning": reasoning,
            "tool_calls": tool_calls,
            "finish_reason": finish_reason,
        }

    @staticmethod
    def _streamed_tool_calls_to_list(
        tool_calls_by_index: dict[int, dict[str, Any]],
    ) -> list[dict[str, Any]] | None:
        """Convert streamed tool_calls index map to list of {id, name, arguments}."""
        if not tool_calls_by_index:
            return None
        result = []
        for idx in sorted(tool_calls_by_index.keys()):
            t = tool_calls_by_index[idx]
            args_str = t.get("arguments") or "{}"
            try:
                arguments = json.loads(args_str)
            except json.JSONDecodeError:
                arguments = {}
            result.append(
                {
                    "id": t.get("id", ""),
                    "name": t.get("name", ""),
                    "arguments": arguments,
                }
            )
        return result if result else None

    def _refresh_media_marker(self) -> bool:
        """Re-fetch /props and update the cached media marker if it changed.

        The server randomizes the marker per startup (unless LLAMA_MEDIA_MARKER
        is set), so a stale marker indicates a restart. Returns True iff the
        marker was updated to a new value — used to gate a one-shot retry of
        a failed embeddings request.
        """
        if self.provider is None:
            return False
        try:
            props = self._fetch_llama_props(self.provider, self.genai_config.model)
        except Exception as e:
            logger.warning("Failed to refresh llama.cpp media marker: %s", e)
            return False

        marker = props.get("media_marker")

        if not isinstance(marker, str) or not marker or marker == self._media_marker:
            return False

        logger.info("llama.cpp media marker changed (server restart); refreshed")
        self._media_marker = marker
        return True

    def embed(
        self,
        texts: list[str] | None = None,
        images: list[bytes] | None = None,
    ) -> list[np.ndarray]:
        """Generate embeddings via llama.cpp /embeddings endpoint.

        Supports batch requests. Uses content format with prompt_string and
        multimodal_data for images (PR #15108). Server must be started with
        --embeddings and --mmproj for multimodal support.
        """
        if self.provider is None:
            logger.warning(
                "llama.cpp provider has not been initialized. Check your llama.cpp configuration."
            )
            return []

        texts = texts or []
        images = images or []
        if not texts and not images:
            return []

        EMBEDDING_DIM = 768

        encoded_images: list[str] = []
        for img in images:
            # llama.cpp uses STB which does not support WebP; convert to JPEG
            jpeg_bytes = _to_jpeg(img)
            to_encode = jpeg_bytes if jpeg_bytes is not None else img
            encoded_images.append(base64.b64encode(to_encode).decode("utf-8"))

        def build_content() -> list[dict[str, Any]]:
            # prompt_string must contain the server's media marker placeholder
            # for each image. The marker is randomized per server startup.
            content: list[dict[str, Any]] = []
            for text in texts:
                content.append({"prompt_string": text})
            for encoded in encoded_images:
                content.append(
                    {
                        "prompt_string": f"{self._media_marker}\n",
                        "multimodal_data": [encoded],
                    }
                )
            return content

        def post_embeddings() -> requests.Response:
            return self._post(
                f"{self.provider}/embeddings",
                json={"model": self.genai_config.model, "content": build_content()},
                timeout=self.timeout,
            )

        try:
            try:
                response = post_embeddings()
                response.raise_for_status()
            except requests.exceptions.RequestException:
                # The server may have restarted with a new media marker.
                # Refresh from /props; only retry if the marker actually changed.
                if not encoded_images or not self._refresh_media_marker():
                    raise
                response = post_embeddings()
                response.raise_for_status()
            result = response.json()

            items = result.get("data", result) if isinstance(result, dict) else result
            if not isinstance(items, list):
                logger.warning("llama.cpp embeddings returned unexpected format")
                return []

            embeddings = []
            for item in items:
                emb = item.get("embedding") if isinstance(item, dict) else None
                if emb is None:
                    logger.warning("llama.cpp embeddings item missing embedding field")
                    continue
                arr = np.array(emb, dtype=np.float32)
                if arr.ndim > 1:
                    # llama.cpp can return token-level embeddings; pool per item
                    arr = arr.mean(axis=0)
                arr = arr.flatten()
                orig_dim = arr.size
                if orig_dim != EMBEDDING_DIM:
                    if orig_dim > EMBEDDING_DIM:
                        arr = arr[:EMBEDDING_DIM]
                        logger.debug(
                            "Truncated llama.cpp embedding from %d to %d dimensions",
                            orig_dim,
                            EMBEDDING_DIM,
                        )
                    else:
                        arr = np.pad(
                            arr,
                            (0, EMBEDDING_DIM - orig_dim),
                            mode="constant",
                            constant_values=0,
                        )
                        logger.debug(
                            "Padded llama.cpp embedding from %d to %d dimensions",
                            orig_dim,
                            EMBEDDING_DIM,
                        )
                embeddings.append(arr)
            return embeddings
        except requests.exceptions.Timeout:
            logger.warning("llama.cpp embeddings request timed out")
            return []
        except requests.exceptions.RequestException as e:
            error_detail = str(e)
            if hasattr(e, "response") and e.response is not None:
                try:
                    error_detail = f"{str(e)} - Response: {e.response.text[:500]}"
                except Exception:
                    pass
            logger.warning("llama.cpp embeddings error: %s", error_detail)
            return []
        except Exception as e:
            logger.warning("Unexpected error in llama.cpp embeddings: %s", str(e))
            return []

    def chat_with_tools(
        self,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]] | None = None,
        tool_choice: str | None = "auto",
        enable_thinking: bool | None = None,
    ) -> dict[str, Any]:
        """
        Send chat messages to llama.cpp server with optional tool definitions.

        Uses the OpenAI-compatible endpoint but passes through all native llama.cpp
        parameters (like slot_id, temperature, etc.) via provider_options.
        """
        if self.provider is None:
            logger.warning(
                "llama.cpp provider has not been initialized. Check your llama.cpp configuration."
            )
            return {
                "content": None,
                "tool_calls": None,
                "finish_reason": "error",
            }
        try:
            payload = self._build_payload(
                messages,
                tools,
                tool_choice,
                stream=False,
                enable_thinking=enable_thinking,
            )
            response = self._post(
                f"{self.provider}/v1/chat/completions",
                json=payload,
                timeout=self.timeout,
            )
            response.raise_for_status()
            result = response.json()
            if result is None or "choices" not in result or len(result["choices"]) == 0:
                return {
                    "content": None,
                    "tool_calls": None,
                    "finish_reason": "error",
                }
            return self._message_from_choice(result["choices"][0])
        except requests.exceptions.Timeout as e:
            logger.warning("llama.cpp request timed out: %s", str(e))
            return {
                "content": None,
                "tool_calls": None,
                "finish_reason": "error",
            }
        except requests.exceptions.RequestException as e:
            error_detail = str(e)
            if hasattr(e, "response") and e.response is not None:
                try:
                    error_detail = f"{str(e)} - Response: {e.response.text[:500]}"
                except Exception:
                    pass
            logger.warning("llama.cpp returned an error: %s", error_detail)
            return {
                "content": None,
                "tool_calls": None,
                "finish_reason": "error",
            }
        except Exception as e:
            logger.warning("Unexpected error in llama.cpp chat_with_tools: %s", str(e))
            return {
                "content": None,
                "tool_calls": None,
                "finish_reason": "error",
            }

    async def chat_with_tools_stream(
        self,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]] | None = None,
        tool_choice: str | None = "auto",
        enable_thinking: bool | None = None,
    ) -> AsyncGenerator[tuple[str, Any], None]:
        """Stream chat with tools via OpenAI-compatible streaming API."""
        if self.provider is None:
            logger.warning(
                "llama.cpp provider has not been initialized. Check your llama.cpp configuration."
            )
            yield (
                "message",
                {
                    "content": None,
                    "tool_calls": None,
                    "finish_reason": "error",
                },
            )
            return
        try:
            payload = self._build_payload(
                messages,
                tools,
                tool_choice,
                stream=True,
                enable_thinking=enable_thinking,
            )
            content_parts: list[str] = []
            reasoning_parts: list[str] = []
            tool_calls_by_index: dict[int, dict[str, Any]] = {}
            finish_reason = "stop"

            async with httpx.AsyncClient(timeout=float(self.timeout)) as client:
                async with client.stream(
                    "POST",
                    f"{self.provider}/v1/chat/completions",
                    json=payload,
                    headers=self._auth_headers(),
                ) as response:
                    response.raise_for_status()
                    async for line in response.aiter_lines():
                        if not line.startswith("data: "):
                            continue
                        data_str = line[6:].strip()
                        if data_str == "[DONE]":
                            break
                        try:
                            data = json.loads(data_str)
                        except json.JSONDecodeError:
                            continue
                        maybe_stats = _stats_from_llama_cpp_chunk(data)
                        if maybe_stats is not None:
                            yield ("stats", maybe_stats)
                        choices = data.get("choices") or []
                        if not choices:
                            continue
                        delta = choices[0].get("delta", {})
                        if choices[0].get("finish_reason"):
                            finish_reason = choices[0]["finish_reason"]
                        # llama.cpp emits separated thinking under
                        # reasoning_content (preferred) or thinking before any
                        # content tokens arrive
                        reasoning_delta = delta.get("reasoning_content") or delta.get(
                            "thinking"
                        )
                        if reasoning_delta:
                            reasoning_parts.append(reasoning_delta)
                            yield ("reasoning_delta", reasoning_delta)
                        if delta.get("content"):
                            content_parts.append(delta["content"])
                            yield ("content_delta", delta["content"])
                        for tc in delta.get("tool_calls") or []:
                            idx = tc.get("index", 0)
                            fn = tc.get("function") or {}
                            if idx not in tool_calls_by_index:
                                tool_calls_by_index[idx] = {
                                    "id": tc.get("id", ""),
                                    "name": tc.get("name") or fn.get("name", ""),
                                    "arguments": "",
                                }
                            t = tool_calls_by_index[idx]
                            if tc.get("id"):
                                t["id"] = tc["id"]
                            name = tc.get("name") or fn.get("name")
                            if name:
                                t["name"] = name
                            arg = tc.get("arguments") or fn.get("arguments")
                            if arg is not None:
                                t["arguments"] += (
                                    arg if isinstance(arg, str) else json.dumps(arg)
                                )

            full_content = "".join(content_parts).strip() or None
            full_reasoning = "".join(reasoning_parts).strip() or None
            tool_calls_list = self._streamed_tool_calls_to_list(tool_calls_by_index)
            if tool_calls_list:
                finish_reason = "tool_calls"
            yield (
                "message",
                {
                    "content": full_content,
                    "reasoning": full_reasoning,
                    "tool_calls": tool_calls_list,
                    "finish_reason": finish_reason,
                },
            )
        except httpx.HTTPStatusError as e:
            logger.warning("llama.cpp streaming HTTP error: %s", e)
            yield (
                "message",
                {
                    "content": None,
                    "tool_calls": None,
                    "finish_reason": "error",
                },
            )
        except Exception as e:
            logger.warning(
                "Unexpected error in llama.cpp chat_with_tools_stream: %s", str(e)
            )
            yield (
                "message",
                {
                    "content": None,
                    "tool_calls": None,
                    "finish_reason": "error",
                },
            )
