import json
from collections.abc import AsyncIterator, Callable, Iterator, Mapping, Sequence
from types import MappingProxyType
from typing import Any, Final, TypeAlias, cast

from typing_extensions import ReadOnly, TypedDict

from litellm import verbose_logger
from litellm.litellm_core_utils.json_validation_rule import normalize_tool_schema
from litellm.types.llms.openai import (
    AllMessageValues,
    ChatCompletionAssistantMessage,
    ChatCompletionAssistantToolCall,
    ChatCompletionImageObject,
    ChatCompletionSystemMessage,
    ChatCompletionTextObject,
    ChatCompletionToolCallFunctionChunk,
    ChatCompletionToolChoiceValues,
    ChatCompletionToolMessage,
    ChatCompletionToolParam,
    ChatCompletionUserMessage,
)
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import (
    AdapterCompletionStreamWrapper,
    ChatCompletionDeltaCustomToolCall,
    ChatCompletionDeltaToolCall,
    ChatCompletionMessageCustomToolCall,
    ChatCompletionMessageToolCall,
    Choices,
    Delta,
    Function,
    Message,
    ModelResponse,
    ModelResponseStream,
    StreamingChoices,
)

_JsonDict: TypeAlias = dict[str, object]
_JsonDictList: TypeAlias = list[_JsonDict]


class _ToolCallAccumulator(TypedDict):
    name: ReadOnly[str]
    arguments: ReadOnly[str]


class _GenAIFunctionCall(TypedDict):
    name: ReadOnly[str]
    args: ReadOnly[Mapping[str, object]]


class _GenAIPart(TypedDict, total=False):
    text: ReadOnly[str]
    functionCall: ReadOnly[_GenAIFunctionCall]


class _GenAIFunctionResponse(TypedDict, total=False):
    name: ReadOnly[str]
    response: ReadOnly[object]


class _GenAIRequestFunctionCall(TypedDict, total=False):
    name: ReadOnly[str]
    args: ReadOnly[Mapping[str, object]]


class _GenAIContentPart(TypedDict, total=False):
    text: ReadOnly[str]
    inline_data: ReadOnly[Mapping[str, str]]
    functionResponse: ReadOnly[_GenAIFunctionResponse]
    functionCall: ReadOnly[_GenAIRequestFunctionCall]


class _GenAIFunctionDeclaration(TypedDict, total=False):
    name: ReadOnly[str]
    description: ReadOnly[str]
    parametersJsonSchema: ReadOnly[object]


class _GenAITool(TypedDict, total=False):
    functionDeclarations: ReadOnly[Sequence[_GenAIFunctionDeclaration]]


class _GenAIFunctionCallingConfig(TypedDict, total=False):
    mode: ReadOnly[str]


class _GenAIToolConfig(TypedDict, total=False):
    functionCallingConfig: ReadOnly[_GenAIFunctionCallingConfig]


class _GenAISystemInstruction(TypedDict, total=False):
    parts: ReadOnly[Sequence[Mapping[str, str]]]


_EMPTY_STR_MAPPING: Final[Mapping[str, str]] = MappingProxyType({})


class GoogleGenAIStreamWrapper(AdapterCompletionStreamWrapper):
    """
    Wrapper for streaming Google GenAI generate_content responses.
    Transforms OpenAI streaming chunks to Google GenAI format.
    """

    sent_first_chunk: bool = False
    _parse_accumulated_args: Callable[[str], Mapping[str, object]] = staticmethod(json.loads)

    def __init__(self, completion_stream: object):
        self.sent_first_chunk = False
        self.accumulated_tool_calls = dict[int, _ToolCallAccumulator]()
        self._returned_response = False
        super().__init__(completion_stream)

    def __next__(self):
        try:
            if not hasattr(self.completion_stream, "__iter__"):
                if self._returned_response:
                    raise StopIteration
                self._returned_response = True
                return GoogleGenAIAdapter().translate_completion_to_generate_content(self.completion_stream)

            for chunk in self.completion_stream:
                if chunk == "None" or chunk is None:
                    continue

                transformed_chunk = GoogleGenAIAdapter().translate_streaming_completion_to_generate_content(chunk, self)
                if transformed_chunk:
                    return transformed_chunk

            raise StopIteration
        except StopIteration:
            raise
        except Exception:
            raise StopIteration

    async def __anext__(self):
        try:
            if not hasattr(self.completion_stream, "__aiter__"):
                if self._returned_response:
                    raise StopAsyncIteration
                self._returned_response = True
                return GoogleGenAIAdapter().translate_completion_to_generate_content(self.completion_stream)

            async for chunk in self.completion_stream:
                if chunk == "None" or chunk is None:
                    continue

                transformed_chunk = GoogleGenAIAdapter().translate_streaming_completion_to_generate_content(chunk, self)
                if transformed_chunk:
                    return transformed_chunk

            # After the stream is exhausted, check for any remaining accumulated tool calls
            if self.accumulated_tool_calls:
                try:
                    parts: Final = list[_GenAIPart]()
                    for (
                        tool_call_index,
                        tool_call_data,
                    ) in self.accumulated_tool_calls.items():
                        try:
                            # For tool calls with no arguments, accumulated_args will be "", which is not valid JSON.
                            # We default to an empty JSON object in this case.
                            parsed_args: Mapping[str, object] = self._parse_accumulated_args(
                                tool_call_data["arguments"] or "{}"
                            )
                            function_call_part: _GenAIPart = {
                                "functionCall": {
                                    "name": tool_call_data["name"] or "undefined_tool_name",
                                    "args": parsed_args,
                                }
                            }
                            parts.append(function_call_part)
                        except json.JSONDecodeError:
                            # This can happen if the stream is abruptly cut off mid-argument string.
                            verbose_logger.warning(
                                "Could not parse tool call arguments at end of stream for index %s. Name: %s. Partial args: %s",
                                tool_call_index,
                                tool_call_data["name"],
                                tool_call_data["arguments"],
                            )
                    if parts:
                        final_chunk: Final = {
                            "candidates": [
                                {
                                    "content": {"parts": parts, "role": "model"},
                                    "finishReason": "STOP",
                                    "index": 0,
                                    "safetyRatings": [],
                                }
                            ]
                        }
                        return final_chunk
                finally:
                    # Ensure the accumulator is always cleared to prevent memory leaks
                    self.accumulated_tool_calls.clear()
            raise StopAsyncIteration
        except StopAsyncIteration:
            raise
        except Exception:
            raise StopAsyncIteration

    def google_genai_sse_wrapper(self) -> Iterator[bytes]:
        """
        Convert Google GenAI streaming chunks to Server-Sent Events format.
        """
        for chunk in self.completion_stream:
            if isinstance(chunk, dict):
                payload = f"data: {json.dumps(chunk)}\n\n"
                yield payload.encode()
            else:
                yield chunk

    async def async_google_genai_sse_wrapper(self) -> AsyncIterator[bytes]:
        """
        Async version of google_genai_sse_wrapper.
        """
        from litellm.types.utils import ModelResponseStream

        async for chunk in self.completion_stream:
            if isinstance(chunk, dict):
                payload = f"data: {json.dumps(chunk)}\n\n"
                yield payload.encode()
            elif isinstance(chunk, ModelResponseStream):
                # Transform OpenAI streaming chunk to Google GenAI format
                transformed_chunk = GoogleGenAIAdapter().translate_streaming_completion_to_generate_content(chunk, self)

                if isinstance(transformed_chunk, dict):  # Only return non-empty chunks
                    payload = f"data: {json.dumps(transformed_chunk)}\n\n"
                    yield payload.encode()
                else:
                    # For empty chunks, continue to next iteration
                    continue
            else:
                # For other chunk types, yield them directly
                if hasattr(chunk, "encode"):
                    yield chunk.encode()
                else:
                    yield str(chunk).encode()


class GoogleGenAIAdapter:
    """Adapter for transforming Google GenAI generate_content requests to/from litellm.completion format"""

    _parse_tool_call_args: Callable[[str], Mapping[str, object]] = staticmethod(json.loads)

    def __init__(self) -> None:
        pass

    def translate_generate_content_to_completion(
        self,
        model: str,
        contents: _JsonDictList | _JsonDict,
        config: Mapping[str, object] | None = None,
        litellm_params: GenericLiteLLMParams | None = None,
        **kwargs,
    ) -> dict[str, Any]:
        """
        Transform generate_content request to litellm completion format

        Args:
            model: The model name
            contents: Generate content contents (can be list or single dict)
            config: Optional config parameters
            **kwargs: Additional parameters from the original request

        Returns:
            Dict in OpenAI format
        """

        # Extract top-level fields from kwargs
        system_instruction: Final = kwargs.get("systemInstruction") or kwargs.get("system_instruction")
        tools: Final = kwargs.get("tools")
        tool_config: Final = kwargs.get("toolConfig") or kwargs.get("tool_config")

        # Normalize contents to list format
        if isinstance(contents, dict):
            contents_list = [contents]
        else:
            contents_list = contents

        # Transform contents to OpenAI messages format
        messages: Final = self._transform_contents_to_messages(contents_list, system_instruction=system_instruction)

        # Create base request as dict (which is compatible with ChatCompletionRequest)
        completion_request: Final[_JsonDict] = {
            "model": model,
            "messages": messages,
        }

        #########################################################
        # Supported OpenAI chat completion params
        # - temperature
        # - max_tokens
        # - top_p
        # - frequency_penalty
        # - presence_penalty
        # - stop
        # - tools
        # - tool_choice
        #########################################################

        # Add config parameters if provided
        if config:
            # Map common Google GenAI config parameters to OpenAI equivalents
            if "temperature" in config:
                completion_request["temperature"] = config["temperature"]
            if "maxOutputTokens" in config:
                completion_request["max_tokens"] = config["maxOutputTokens"]
            if "topP" in config:
                completion_request["top_p"] = config["topP"]
            if "topK" in config:
                # OpenAI doesn't have direct topK, but we can pass it as extra
                pass
            if "stopSequences" in config:
                completion_request["stop"] = config["stopSequences"]

        # Handle tools transformation
        if tools:
            # Check if tools are already in OpenAI format or Google GenAI format
            if isinstance(tools, list) and len(tools) > 0:
                # Tools are in Google GenAI format, transform them
                openai_tools: Final = self._transform_google_genai_tools_to_openai(tools)

                if openai_tools:
                    completion_request["tools"] = openai_tools

        # Handle tool_config (tool choice)
        if tool_config:
            tool_choice: Final = self._transform_google_genai_tool_config_to_openai(tool_config)
            if tool_choice:
                completion_request["tool_choice"] = tool_choice

        #########################################################
        # forward any litellm specific params
        #########################################################
        completion_request_dict = dict(completion_request)
        if litellm_params:
            completion_request_dict = self._add_generic_litellm_params_to_request(
                completion_request_dict=completion_request_dict,
                litellm_params=litellm_params,
            )

        return completion_request_dict

    def _add_generic_litellm_params_to_request(
        self,
        completion_request_dict: _JsonDict,
        litellm_params: GenericLiteLLMParams | None = None,
    ) -> _JsonDict:
        """Add generic litellm params to request. e.g add api_base, api_key, api_version, etc.

        Args:
            completion_request_dict: Dict[str, Any]
            litellm_params: GenericLiteLLMParams

        Returns:
            Dict[str, Any]
        """
        allowed_fields: Final = GenericLiteLLMParams.model_fields.keys()
        if litellm_params:
            litellm_dict: Final[_JsonDict] = litellm_params.model_dump(exclude_none=True)
            for key, value in litellm_dict.items():
                if key in allowed_fields:
                    completion_request_dict[key] = value
        return completion_request_dict

    def translate_completion_output_params_streaming(
        self,
        completion_stream: object,
    ) -> AsyncIterator[bytes] | None:
        """Transform streaming completion output to Google GenAI format"""
        google_genai_wrapper: Final = GoogleGenAIStreamWrapper(completion_stream=completion_stream)
        # Return the SSE-wrapped version for proper event formatting
        return google_genai_wrapper.async_google_genai_sse_wrapper()

    def _transform_google_genai_tools_to_openai(
        self,
        tools: Sequence[_GenAITool],
    ) -> list[ChatCompletionToolParam]:
        """Transform Google GenAI tools to OpenAI tools format"""
        openai_tools: Final = list[_JsonDict]()

        for tool in tools:
            if "functionDeclarations" in tool:
                for func_decl in tool["functionDeclarations"]:
                    function_chunk: _JsonDict = {
                        "name": func_decl.get("name", ""),
                    }

                    if "description" in func_decl:
                        function_chunk["description"] = func_decl["description"]
                    if "parametersJsonSchema" in func_decl:
                        function_chunk["parameters"] = func_decl["parametersJsonSchema"]

                    openai_tool: _JsonDict = {"type": "function", "function": function_chunk}
                    openai_tools.append(openai_tool)

        # normalize the tool schemas
        normalized_tools: Final = [normalize_tool_schema(tool) for tool in openai_tools]

        return cast(list[ChatCompletionToolParam], normalized_tools)

    def _transform_google_genai_tool_config_to_openai(
        self,
        tool_config: _GenAIToolConfig,
    ) -> ChatCompletionToolChoiceValues | None:
        """Transform Google GenAI tool_config to OpenAI tool_choice"""
        function_calling_config: Final = tool_config.get("functionCallingConfig", {})
        mode: Final = function_calling_config.get("mode", "AUTO")

        mode_mapping: Final = {"AUTO": "auto", "ANY": "required", "NONE": "none"}

        tool_choice: Final = mode_mapping.get(mode, "auto")
        return cast(ChatCompletionToolChoiceValues, tool_choice)

    def _transform_contents_to_messages(
        self,
        contents: list[dict[str, Any]],
        system_instruction: _GenAISystemInstruction | None = None,
    ) -> list[AllMessageValues]:
        """Transform Google GenAI contents to OpenAI messages format"""
        messages: Final[list[AllMessageValues]] = []

        # Handle system instruction
        if system_instruction:
            system_parts: Final[Sequence[Mapping[str, str]]] = system_instruction.get("parts", [])
            if system_parts and "text" in system_parts[0]:
                messages.append(ChatCompletionSystemMessage(role="system", content=system_parts[0]["text"]))

        for content in contents:
            role = content.get("role", "user")
            parts: Sequence[_GenAIContentPart | str | None] = content.get("parts", [])

            if role == "user":
                # Handle user messages with potential function responses
                content_parts: list[ChatCompletionTextObject | ChatCompletionImageObject] = []
                tool_messages: list[ChatCompletionToolMessage] = []

                for part in parts:
                    if isinstance(part, dict):
                        if "text" in part:
                            content_parts.append(
                                cast(
                                    ChatCompletionTextObject,
                                    {"type": "text", "text": part["text"]},
                                )
                            )
                        elif "inline_data" in part:
                            # Handle Base64 image data
                            inline_data = part["inline_data"]
                            mime_type = inline_data.get("mime_type", "image/jpeg")
                            data = inline_data.get("data", "")
                            content_parts.append(
                                cast(
                                    ChatCompletionImageObject,
                                    {
                                        "type": "image_url",
                                        "image_url": {"url": f"data:{mime_type};base64,{data}"},
                                    },
                                )
                            )
                        elif "functionResponse" in part:
                            # Transform function response to tool message
                            func_response = part["functionResponse"]
                            tool_message = ChatCompletionToolMessage(
                                role="tool",
                                tool_call_id=f"call_{func_response.get('name', 'unknown')}",
                                content=json.dumps(func_response.get("response", {})),
                            )
                            tool_messages.append(tool_message)
                    elif isinstance(part, str):
                        content_parts.append(cast(ChatCompletionTextObject, {"type": "text", "text": part}))

                # Add user message if there's content
                if content_parts:
                    # If only one text part, use simple string format for backward compatibility
                    if (
                        len(content_parts) == 1
                        and isinstance(content_parts[0], dict)
                        and content_parts[0].get("type") == "text"
                    ):
                        text_part = cast(ChatCompletionTextObject, content_parts[0])
                        messages.append(ChatCompletionUserMessage(role="user", content=text_part["text"]))
                    else:
                        # Use multimodal format (array of content parts)
                        messages.append(ChatCompletionUserMessage(role="user", content=content_parts))

                # Add tool messages
                messages.extend(tool_messages)

            elif role == "model":
                # Handle assistant messages with potential function calls
                combined_text = ""
                tool_calls: list[ChatCompletionAssistantToolCall] = []

                for part in parts:
                    if isinstance(part, dict):
                        if "text" in part:
                            combined_text += part["text"]
                        elif "functionCall" in part:
                            # Transform function call to tool call
                            func_call = part["functionCall"]
                            tool_call = ChatCompletionAssistantToolCall(
                                id=f"call_{func_call.get('name', 'unknown')}",
                                type="function",
                                function=ChatCompletionToolCallFunctionChunk(
                                    name=func_call.get("name", ""),
                                    arguments=json.dumps(func_call.get("args", {})),
                                ),
                            )
                            tool_calls.append(tool_call)
                    elif isinstance(part, str):
                        combined_text += part

                # Create assistant message
                if tool_calls:
                    assistant_message = ChatCompletionAssistantMessage(
                        role="assistant",
                        content=combined_text if combined_text else None,
                        tool_calls=tool_calls,
                    )
                else:
                    assistant_message = ChatCompletionAssistantMessage(
                        role="assistant",
                        content=combined_text if combined_text else None,
                    )

                messages.append(assistant_message)

        return messages

    def translate_completion_to_generate_content(
        self,
        response: ModelResponse,
    ) -> _JsonDict:
        """
        Transform litellm completion response to Google GenAI generate_content format

        Args:
            response: ModelResponse from litellm.completion

        Returns:
            Dict in Google GenAI generate_content response format
        """

        # Extract the main response content
        choice: Final = response.choices[0] if response.choices else None
        if not choice:
            raise ValueError("Invalid completion response: no choices found")

        # Handle different choice types (Choices vs StreamingChoices)
        if isinstance(choice, Choices):
            if not choice.message:
                raise ValueError("Invalid completion response: no message found in choice")
            parts = self._transform_openai_message_to_google_genai_parts(choice.message)
        else:
            # Fallback for generic choice objects
            message_content: str = getattr(choice, "message", _EMPTY_STR_MAPPING).get("content", "") or getattr(
                choice, "delta", _EMPTY_STR_MAPPING
            ).get("content", "")
            parts = [{"text": message_content}] if message_content else []

        # Create Google GenAI format response
        generate_content_response: Final[_JsonDict] = {
            "candidates": [
                {
                    "content": {"parts": parts, "role": "model"},
                    "finishReason": self._map_finish_reason(getattr(choice, "finish_reason", None)),
                    "index": 0,
                    "safetyRatings": [],
                }
            ],
            "usageMetadata": (
                self._map_usage(getattr(response, "usage", None))
                if hasattr(response, "usage") and getattr(response, "usage", None)
                else {
                    "promptTokenCount": 0,
                    "candidatesTokenCount": 0,
                    "totalTokenCount": 0,
                }
            ),
        }

        # Add text field for convenience (common in Google GenAI responses)
        text_content = ""
        for part in parts:
            if isinstance(part, dict) and "text" in part:
                text_content += part["text"]
        if text_content:
            generate_content_response["text"] = text_content

        return generate_content_response

    def translate_streaming_completion_to_generate_content(
        self,
        response: ModelResponse | ModelResponseStream,
        wrapper: GoogleGenAIStreamWrapper,
    ) -> Mapping[str, object] | None:
        """
        Transform streaming litellm completion chunk to Google GenAI generate_content format

        Args:
            response: Streaming ModelResponse chunk from litellm.completion
            wrapper: GoogleGenAIStreamWrapper instance

        Returns:
            Dict in Google GenAI streaming generate_content response format
        """

        # Extract the main response content from streaming chunk
        choice: Final = response.choices[0] if response.choices else None
        if not choice:
            # Return empty chunk if no choices
            return None

        # Handle streaming choice
        if isinstance(choice, StreamingChoices):
            if choice.delta:
                parts = self._transform_openai_delta_to_google_genai_parts_with_accumulation(choice.delta, wrapper)
            else:
                parts = []
            finish_reason: str | None = getattr(choice, "finish_reason", None)
        else:
            # Fallback for generic choice objects
            message_content: Final[str] = getattr(choice, "delta", _EMPTY_STR_MAPPING).get("content", "")
            parts = [{"text": message_content}] if message_content else []
            finish_reason = getattr(choice, "finish_reason", None)

        # Only create response chunk if we have parts or it's the final chunk
        if not parts and not finish_reason:
            return None

        # Create Google GenAI streaming format response
        streaming_chunk: Final[_JsonDict] = {
            "candidates": [
                {
                    "content": {"parts": parts, "role": "model"},
                    "finishReason": (self._map_finish_reason(finish_reason) if finish_reason else None),
                    "index": 0,
                    "safetyRatings": [],
                }
            ]
        }

        # Add usage metadata only in the final chunk (when finish_reason is present)
        if finish_reason:
            usage_metadata: Final = (
                self._map_usage(getattr(response, "usage", None))
                if hasattr(response, "usage") and getattr(response, "usage", None)
                else {
                    "promptTokenCount": 0,
                    "candidatesTokenCount": 0,
                    "totalTokenCount": 0,
                }
            )
            streaming_chunk["usageMetadata"] = usage_metadata

        # Add text field for convenience (common in Google GenAI responses)
        text_content = ""
        for part in parts:
            if isinstance(part, dict) and "text" in part:
                text_content += part["text"]
        if text_content:
            streaming_chunk["text"] = text_content

        return streaming_chunk

    def _transform_openai_message_to_google_genai_parts(
        self,
        message: Message,
    ) -> Sequence[_GenAIPart]:
        """Transform OpenAI message to Google GenAI parts format"""
        parts: Final = list[_GenAIPart]()

        # Add text content if present
        if hasattr(message, "content") and message.content:
            parts.append({"text": message.content})

        # Add tool calls if present
        if hasattr(message, "tool_calls") and message.tool_calls:
            tool_calls: Final[Sequence[ChatCompletionMessageToolCall | ChatCompletionMessageCustomToolCall]] = (
                message.tool_calls
            )
            for tool_call in tool_calls:
                function: Function | None = getattr(tool_call, "function", None)
                if function:
                    try:
                        args: Mapping[str, object] = (
                            self._parse_tool_call_args(function.arguments) if function.arguments else {}
                        )
                    except json.JSONDecodeError:
                        args = {}

                    function_call_part: _GenAIPart = {
                        "functionCall": {
                            "name": function.name or "undefined_tool_name",
                            "args": args,
                        }
                    }
                    parts.append(function_call_part)

        return parts if parts else [{"text": ""}]

    def _transform_openai_delta_to_google_genai_parts_with_accumulation(
        self, delta: Delta, wrapper: GoogleGenAIStreamWrapper
    ) -> Sequence[_GenAIPart]:
        """Transforms OpenAI delta to Google GenAI parts, accumulating streaming tool calls."""

        # 1. Initialize wrapper state if it doesn't exist
        if not hasattr(wrapper, "accumulated_tool_calls"):
            wrapper.accumulated_tool_calls = {}

        parts: Final = list[_GenAIPart]()

        if hasattr(delta, "content") and delta.content:
            parts.append({"text": delta.content})

        # 2. Ensure tool_calls is iterable
        tool_calls: Final[Sequence[ChatCompletionDeltaToolCall | ChatCompletionDeltaCustomToolCall]] = (
            delta.tool_calls or []
        )

        for tool_call in tool_calls:
            if not hasattr(tool_call, "function") or isinstance(tool_call, ChatCompletionDeltaCustomToolCall):
                continue

            # 3. Use `index` as the primary key for accumulation
            tool_call_index: int | None = getattr(tool_call, "index", None)
            if tool_call_index is None:
                continue  # Index is essential for tracking streaming tool calls

            # Initialize accumulator for this index if it's new
            if tool_call_index not in wrapper.accumulated_tool_calls:
                wrapper.accumulated_tool_calls[tool_call_index] = {
                    "name": "",
                    "arguments": "",
                }

            # Accumulate name and arguments
            delta_function: Function | None = getattr(tool_call, "function", None)
            function_name: str | None = getattr(delta_function, "name", None)
            args_chunk: str | None = getattr(delta_function, "arguments", None)

            # Optimization: Skip chunks that have no new data
            if not function_name and not args_chunk:
                verbose_logger.debug("Skipping empty tool call chunk for index: %s", tool_call_index)
                continue

            previous_data: _ToolCallAccumulator = wrapper.accumulated_tool_calls[tool_call_index]
            wrapper.accumulated_tool_calls[tool_call_index] = _ToolCallAccumulator(
                name=function_name or previous_data["name"],
                arguments=previous_data["arguments"] + (args_chunk or ""),
            )

            # Attempt to parse and emit a complete tool call
            accumulated_data = wrapper.accumulated_tool_calls[tool_call_index]
            accumulated_name = accumulated_data["name"]
            accumulated_args = accumulated_data["arguments"]

            # 5. Attempt to parse arguments even if name hasn't arrived.
            try:
                # Attempt to parse the accumulated arguments string
                parsed_args: Mapping[str, object] = self._parse_tool_call_args(accumulated_args)

                # If parsing succeeds, but we don't have a name yet, wait.
                # The part will be created by a later chunk that brings the name.
                if accumulated_name:
                    # If successful, create the part and clean up
                    function_call_part: _GenAIPart = {"functionCall": {"name": accumulated_name, "args": parsed_args}}
                    parts.append(function_call_part)

                    # Remove the completed tool call from the accumulator
                    del wrapper.accumulated_tool_calls[tool_call_index]

            except json.JSONDecodeError:
                # The JSON for arguments is still incomplete.
                # We will continue to accumulate and wait for more chunks.
                pass

        return parts

    def _map_finish_reason(self, finish_reason: str | None) -> str:
        """Map OpenAI finish reasons to Google GenAI finish reasons"""
        if not finish_reason:
            return "STOP"

        mapping: Final = {
            "stop": "STOP",
            "length": "MAX_TOKENS",
            "content_filter": "SAFETY",
            "tool_calls": "STOP",
            "function_call": "STOP",
        }

        return mapping.get(finish_reason, "STOP")

    def _map_usage(self, usage: object) -> Mapping[str, int]:
        """Map OpenAI usage to Google GenAI usage format"""
        return {
            "promptTokenCount": getattr(usage, "prompt_tokens", 0) or 0,
            "candidatesTokenCount": getattr(usage, "completion_tokens", 0) or 0,
            "totalTokenCount": getattr(usage, "total_tokens", 0) or 0,
        }
