"""Derivations of standard content blocks from OpenAI content."""

from __future__ import annotations

import json
import warnings
from typing import TYPE_CHECKING, Any, Literal, cast

from langchain_core.language_models._utils import (
    _parse_data_uri,
    is_openai_data_block,
)
from langchain_core.messages import AIMessageChunk
from langchain_core.messages import content as types

if TYPE_CHECKING:
    from collections.abc import Iterator

    from langchain_core.messages import AIMessage


def convert_to_openai_image_block(block: dict[str, Any]) -> dict[str, Any]:
    """Convert `ImageContentBlock` to format expected by OpenAI Chat Completions.

    Args:
        block: The image content block to convert.

    Raises:
        ValueError: If required keys are missing.
        ValueError: If source type is unsupported.

    Returns:
        The formatted image content block.
    """
    if "url" in block:
        return {
            "type": "image_url",
            "image_url": {
                "url": block["url"],
            },
        }
    if "base64" in block or block.get("source_type") == "base64":
        if "mime_type" not in block:
            error_message = "mime_type key is required for base64 data."
            raise ValueError(error_message)
        mime_type = block["mime_type"]
        base64_data = block["data"] if "data" in block else block["base64"]
        return {
            "type": "image_url",
            "image_url": {
                "url": f"data:{mime_type};base64,{base64_data}",
            },
        }
    error_message = "Unsupported source type. Only 'url' and 'base64' are supported."
    raise ValueError(error_message)


def convert_to_openai_data_block(
    block: dict[str, Any],
    api: Literal["chat/completions", "responses"] = "chat/completions",
) -> dict[str, Any]:
    """Format standard data content block to format expected by OpenAI.

    "Standard data content block" can include old-style LangChain v0 blocks
    (URLContentBlock, Base64ContentBlock, IDContentBlock) or new ones.

    Args:
        block: The content block to convert.
        api: The OpenAI API being targeted. Either "chat/completions" or "responses".

    Raises:
        ValueError: If required keys are missing.
        ValueError: If file URLs are used with Chat Completions API.
        ValueError: If block type is unsupported.

    Returns:
        The formatted content block.
    """
    if block["type"] == "image":
        chat_completions_block = convert_to_openai_image_block(block)
        if api == "responses":
            formatted_block = {
                "type": "input_image",
                "image_url": chat_completions_block["image_url"]["url"],
            }
            if chat_completions_block["image_url"].get("detail"):
                formatted_block["detail"] = chat_completions_block["image_url"][
                    "detail"
                ]
        else:
            formatted_block = chat_completions_block

    elif block["type"] == "file":
        if block.get("source_type") == "base64" or "base64" in block:
            # Handle v0 format (Base64CB): {"source_type": "base64", "data": "...", ...}
            # Handle v1 format (IDCB): {"base64": "...", ...}
            base64_data = block["data"] if "source_type" in block else block["base64"]
            file = {"file_data": f"data:{block['mime_type']};base64,{base64_data}"}
            if filename := block.get("filename"):
                file["filename"] = filename
            elif (extras := block.get("extras")) and ("filename" in extras):
                file["filename"] = extras["filename"]
            elif (extras := block.get("metadata")) and ("filename" in extras):
                # Backward compat
                file["filename"] = extras["filename"]
            else:
                # Can't infer filename; set a placeholder default for compatibility.
                file["filename"] = "LC_AUTOGENERATED"
                warnings.warn(
                    "OpenAI may require a filename for file uploads. Specify a filename"
                    " in the content block, e.g.: {'type': 'file', 'mime_type': "
                    "'...', 'base64': '...', 'filename': 'my-file.pdf'}. "
                    "Using placeholder filename 'LC_AUTOGENERATED'.",
                    stacklevel=1,
                )
            formatted_block = {"type": "file", "file": file}
            if api == "responses":
                formatted_block = {"type": "input_file", **formatted_block["file"]}
        elif block.get("source_type") == "id" or "file_id" in block:
            # Handle v0 format (IDContentBlock): {"source_type": "id", "id": "...", ...}
            # Handle v1 format (IDCB): {"file_id": "...", ...}
            file_id = block["id"] if "source_type" in block else block["file_id"]
            formatted_block = {"type": "file", "file": {"file_id": file_id}}
            if api == "responses":
                formatted_block = {"type": "input_file", **formatted_block["file"]}
        elif "url" in block:  # Intentionally do not check for source_type="url"
            if api == "chat/completions":
                error_msg = "OpenAI Chat Completions does not support file URLs."
                raise ValueError(error_msg)
            # Only supported by Responses API; return in that format
            formatted_block = {"type": "input_file", "file_url": block["url"]}
        else:
            error_msg = "Keys base64, url, or file_id required for file blocks."
            raise ValueError(error_msg)

    elif block["type"] == "audio":
        if "base64" in block or block.get("source_type") == "base64":
            # Handle v0 format: {"source_type": "base64", "data": "...", ...}
            # Handle v1 format: {"base64": "...", ...}
            base64_data = block["data"] if "source_type" in block else block["base64"]
            audio_format = block["mime_type"].split("/")[-1]
            formatted_block = {
                "type": "input_audio",
                "input_audio": {"data": base64_data, "format": audio_format},
            }
        else:
            error_msg = "Key base64 is required for audio blocks."
            raise ValueError(error_msg)
    else:
        error_msg = f"Block of type {block['type']} is not supported."
        raise ValueError(error_msg)

    return formatted_block


# v1 / Chat Completions
def _convert_to_v1_from_chat_completions(
    message: AIMessage,
) -> list[types.ContentBlock]:
    """Mutate a Chat Completions message to v1 format."""
    content_blocks: list[types.ContentBlock] = []
    if isinstance(message.content, str):
        if message.content:
            content_blocks = [{"type": "text", "text": message.content}]
        else:
            content_blocks = []

    for tool_call in message.tool_calls:
        content_blocks.append(
            {
                "type": "tool_call",
                "name": tool_call["name"],
                "args": tool_call["args"],
                "id": tool_call.get("id"),
            }
        )

    return content_blocks


def _convert_to_v1_from_chat_completions_input(
    content: list[types.ContentBlock],
) -> list[types.ContentBlock]:
    """Convert OpenAI Chat Completions format blocks to v1 format.

    During the `content_blocks` parsing process, we wrap blocks not recognized as a v1
    block as a `'non_standard'` block with the original block stored in the `value`
    field. This function attempts to unpack those blocks and convert any blocks that
    might be OpenAI format to v1 ContentBlocks.

    If conversion fails, the block is left as a `'non_standard'` block.

    Args:
        content: List of content blocks to process.

    Returns:
        Updated list with OpenAI blocks converted to v1 format.
    """
    converted_blocks = []
    unpacked_blocks: list[dict[str, Any]] = [
        cast("dict[str, Any]", block)
        if block.get("type") != "non_standard"
        else block["value"]  # type: ignore[typeddict-item]  # this is only non-standard blocks
        for block in content
    ]
    for block in unpacked_blocks:
        if block.get("type") in {
            "image_url",
            "input_audio",
            "file",
        } and is_openai_data_block(block):
            converted_block = _convert_openai_format_to_data_block(block)
            # If conversion succeeded, use it; otherwise keep as non_standard
            if (
                isinstance(converted_block, dict)
                and converted_block.get("type") in types.KNOWN_BLOCK_TYPES
            ):
                converted_blocks.append(cast("types.ContentBlock", converted_block))
            else:
                converted_blocks.append({"type": "non_standard", "value": block})
        elif block.get("type") in types.KNOWN_BLOCK_TYPES:
            converted_blocks.append(cast("types.ContentBlock", block))
        else:
            converted_blocks.append({"type": "non_standard", "value": block})

    return converted_blocks


def _convert_to_v1_from_chat_completions_chunk(
    chunk: AIMessageChunk,
) -> list[types.ContentBlock]:
    """Mutate a Chat Completions chunk to v1 format."""
    content_blocks: list[types.ContentBlock] = []
    if isinstance(chunk.content, str):
        if chunk.content:
            content_blocks = [{"type": "text", "text": chunk.content}]
        else:
            content_blocks = []

    if chunk.chunk_position == "last":
        for tool_call in chunk.tool_calls:
            content_blocks.append(
                {
                    "type": "tool_call",
                    "name": tool_call["name"],
                    "args": tool_call["args"],
                    "id": tool_call.get("id"),
                }
            )

    else:
        for tool_call_chunk in chunk.tool_call_chunks:
            tc: types.ToolCallChunk = {
                "type": "tool_call_chunk",
                "id": tool_call_chunk.get("id"),
                "name": tool_call_chunk.get("name"),
                "args": tool_call_chunk.get("args"),
            }
            if (idx := tool_call_chunk.get("index")) is not None:
                tc["index"] = idx
            content_blocks.append(tc)

    return content_blocks


def _convert_from_v1_to_chat_completions(message: AIMessage) -> AIMessage:
    """Convert a v1 message to the Chat Completions format."""
    if isinstance(message.content, list):
        new_content: list[Any] = []
        for block in message.content:
            if isinstance(block, dict):
                block_type = block.get("type")
                if block_type == "text":
                    # Strip annotations
                    new_content.append({"type": "text", "text": block["text"]})
                elif block_type in {"reasoning", "tool_call"}:
                    pass
                else:
                    new_content.append(block)
            else:
                new_content.append(block)
        return message.model_copy(update={"content": new_content})

    return message


# Responses
_FUNCTION_CALL_IDS_MAP_KEY = "__openai_function_call_ids__"


def _convert_from_v03_ai_message(message: AIMessage) -> AIMessage:
    """Convert v0 AIMessage into `output_version="responses/v1"` format."""
    # Only update ChatOpenAI v0.3 AIMessages
    is_chatopenai_v03 = (
        isinstance(message.content, list)
        and all(isinstance(b, dict) for b in message.content)
    ) and (
        any(
            item in message.additional_kwargs
            for item in [
                "reasoning",
                "tool_outputs",
                "refusal",
                _FUNCTION_CALL_IDS_MAP_KEY,
            ]
        )
        or (
            isinstance(message.id, str)
            and message.id.startswith("msg_")
            and (response_id := message.response_metadata.get("id"))
            and isinstance(response_id, str)
            and response_id.startswith("resp_")
        )
    )
    if not is_chatopenai_v03:
        return message

    content_order = [
        "reasoning",
        "code_interpreter_call",
        "mcp_call",
        "image_generation_call",
        "text",
        "refusal",
        "function_call",
        "computer_call",
        "mcp_list_tools",
        "mcp_approval_request",
        # N. B. "web_search_call" and "file_search_call" were not passed back in
        # in v0.3
    ]

    # Build a bucket for every known block type
    buckets: dict[str, list[Any]] = {key: [] for key in content_order}
    unknown_blocks = []

    # Reasoning
    if reasoning := message.additional_kwargs.get("reasoning"):
        if "type" not in reasoning:
            reasoning = {**reasoning, "type": "reasoning"}
        buckets["reasoning"].append(reasoning)

    # Refusal
    if refusal := message.additional_kwargs.get("refusal"):
        buckets["refusal"].append({"type": "refusal", "refusal": refusal})

    # Text
    for block in message.content:
        if isinstance(block, dict) and block.get("type") == "text":
            block_copy = block.copy()
            if isinstance(message.id, str) and message.id.startswith("msg_"):
                block_copy["id"] = message.id
            buckets["text"].append(block_copy)
        else:
            unknown_blocks.append(block)

    # Function calls
    function_call_ids = message.additional_kwargs.get(_FUNCTION_CALL_IDS_MAP_KEY)
    if (
        isinstance(message, AIMessageChunk)
        and len(message.tool_call_chunks) == 1
        and message.chunk_position != "last"
    ):
        # Isolated chunk
        tool_call_chunk = message.tool_call_chunks[0]
        function_call = {
            "type": "function_call",
            "name": tool_call_chunk.get("name"),
            "arguments": tool_call_chunk.get("args"),
            "call_id": tool_call_chunk.get("id"),
        }
        if function_call_ids is not None and (
            id_ := function_call_ids.get(tool_call_chunk.get("id"))
        ):
            function_call["id"] = id_
        buckets["function_call"].append(function_call)
    else:
        for tool_call in message.tool_calls:
            function_call = {
                "type": "function_call",
                "name": tool_call["name"],
                "arguments": json.dumps(tool_call["args"], ensure_ascii=False),
                "call_id": tool_call["id"],
            }
            if function_call_ids is not None and (
                id_ := function_call_ids.get(tool_call["id"])
            ):
                function_call["id"] = id_
            buckets["function_call"].append(function_call)

    # Tool outputs
    tool_outputs = message.additional_kwargs.get("tool_outputs", [])
    for block in tool_outputs:
        if isinstance(block, dict) and (key := block.get("type")) and key in buckets:
            buckets[key].append(block)
        else:
            unknown_blocks.append(block)

    # Re-assemble the content list in the canonical order
    new_content = []
    for key in content_order:
        new_content.extend(buckets[key])
    new_content.extend(unknown_blocks)

    new_additional_kwargs = dict(message.additional_kwargs)
    new_additional_kwargs.pop("reasoning", None)
    new_additional_kwargs.pop("refusal", None)
    new_additional_kwargs.pop("tool_outputs", None)

    if "id" in message.response_metadata:
        new_id = message.response_metadata["id"]
    else:
        new_id = message.id

    return message.model_copy(
        update={
            "content": new_content,
            "additional_kwargs": new_additional_kwargs,
            "id": new_id,
        },
        deep=False,
    )


def _convert_openai_format_to_data_block(
    block: dict[str, Any],
) -> types.ContentBlock | dict[str, Any]:
    """Convert OpenAI image/audio/file content block to respective v1 multimodal block.

    We expect that the incoming block is verified to be in OpenAI Chat Completions
    format.

    If parsing fails, passes block through unchanged.

    Mappings (Chat Completions to LangChain v1):
    - Image -> `ImageContentBlock`
    - Audio -> `AudioContentBlock`
    - File -> `FileContentBlock`

    """

    # Extract extra keys to put them in `extras`
    def _extract_extras(
        block_dict: dict[str, Any], known_keys: set[str]
    ) -> dict[str, Any]:
        """Extract unknown keys from block to preserve as extras."""
        return {k: v for k, v in block_dict.items() if k not in known_keys}

    # base64-style image block
    if (block["type"] == "image_url") and (
        parsed := _parse_data_uri(block["image_url"]["url"])
    ):
        known_keys = {"type", "image_url"}
        extras = _extract_extras(block, known_keys)

        # Also extract extras from nested image_url dict
        image_url_known_keys = {"url"}
        image_url_extras = _extract_extras(block["image_url"], image_url_known_keys)

        # Merge extras
        all_extras = {**extras}
        for key, value in image_url_extras.items():
            if key == "detail":  # Don't rename
                all_extras["detail"] = value
            else:
                all_extras[f"image_url_{key}"] = value

        return types.create_image_block(
            # Even though this is labeled as `url`, it can be base64-encoded
            base64=parsed["data"],
            mime_type=parsed["mime_type"],
            **all_extras,
        )

    # url-style image block
    if (block["type"] == "image_url") and isinstance(
        block["image_url"].get("url"), str
    ):
        known_keys = {"type", "image_url"}
        extras = _extract_extras(block, known_keys)

        image_url_known_keys = {"url"}
        image_url_extras = _extract_extras(block["image_url"], image_url_known_keys)

        all_extras = {**extras}
        for key, value in image_url_extras.items():
            if key == "detail":  # Don't rename
                all_extras["detail"] = value
            else:
                all_extras[f"image_url_{key}"] = value

        return types.create_image_block(
            url=block["image_url"]["url"],
            **all_extras,
        )

    # base64-style audio block
    # audio is only represented via raw data, no url or ID option
    if block["type"] == "input_audio":
        known_keys = {"type", "input_audio"}
        extras = _extract_extras(block, known_keys)

        # Also extract extras from nested audio dict
        audio_known_keys = {"data", "format"}
        audio_extras = _extract_extras(block["input_audio"], audio_known_keys)

        all_extras = {**extras}
        for key, value in audio_extras.items():
            all_extras[f"audio_{key}"] = value

        return types.create_audio_block(
            base64=block["input_audio"]["data"],
            mime_type=f"audio/{block['input_audio']['format']}",
            **all_extras,
        )

    # id-style file block
    if block.get("type") == "file" and "file_id" in block.get("file", {}):
        known_keys = {"type", "file"}
        extras = _extract_extras(block, known_keys)

        file_known_keys = {"file_id"}
        file_extras = _extract_extras(block["file"], file_known_keys)

        all_extras = {**extras}
        for key, value in file_extras.items():
            all_extras[f"file_{key}"] = value

        return types.create_file_block(
            file_id=block["file"]["file_id"],
            **all_extras,
        )

    # base64-style file block
    if (block["type"] == "file") and (
        parsed := _parse_data_uri(block["file"]["file_data"])
    ):
        known_keys = {"type", "file"}
        extras = _extract_extras(block, known_keys)

        file_known_keys = {"file_data", "filename"}
        file_extras = _extract_extras(block["file"], file_known_keys)

        all_extras = {**extras}
        for key, value in file_extras.items():
            all_extras[f"file_{key}"] = value

        filename = block["file"].get("filename")
        return types.create_file_block(
            base64=parsed["data"],
            mime_type="application/pdf",
            filename=filename,
            **all_extras,
        )

    # Escape hatch
    return block


# v1 / Responses
def _convert_annotation_to_v1(annotation: dict[str, Any]) -> types.Annotation:
    annotation_type = annotation.get("type")

    if annotation_type == "url_citation":
        known_fields = {
            "type",
            "url",
            "title",
            "cited_text",
            "start_index",
            "end_index",
        }
        url_citation = cast("types.Citation", {})
        for field in ("end_index", "start_index", "title"):
            if field in annotation:
                url_citation[field] = annotation[field]
        url_citation["type"] = "citation"
        url_citation["url"] = annotation["url"]
        for field, value in annotation.items():
            if field not in known_fields:
                if "extras" not in url_citation:
                    url_citation["extras"] = {}
                url_citation["extras"][field] = value
        return url_citation

    if annotation_type == "file_citation":
        known_fields = {
            "type",
            "title",
            "cited_text",
            "start_index",
            "end_index",
            "filename",
        }
        document_citation: types.Citation = {"type": "citation"}
        if "filename" in annotation:
            document_citation["title"] = annotation["filename"]
        for field, value in annotation.items():
            if field not in known_fields:
                if "extras" not in document_citation:
                    document_citation["extras"] = {}
                document_citation["extras"][field] = value

        return document_citation

    # TODO: standardise container_file_citation?
    non_standard_annotation: types.NonStandardAnnotation = {
        "type": "non_standard_annotation",
        "value": annotation,
    }
    return non_standard_annotation


def _explode_reasoning(block: dict[str, Any]) -> Iterator[types.ReasoningContentBlock]:
    if "summary" not in block:
        yield cast("types.ReasoningContentBlock", block)
        return

    known_fields = {"type", "reasoning", "id", "index"}
    unknown_fields = [
        field for field in block if field != "summary" and field not in known_fields
    ]
    if unknown_fields:
        block["extras"] = {}
    for field in unknown_fields:
        block["extras"][field] = block.pop(field)

    if not block["summary"]:
        # [{'id': 'rs_...', 'summary': [], 'type': 'reasoning', 'index': 0}]
        block = {k: v for k, v in block.items() if k != "summary"}
        if "index" in block:
            meaningful_idx = f"{block['index']}_0"
            block["index"] = f"lc_rs_{meaningful_idx.encode().hex()}"
        yield cast("types.ReasoningContentBlock", block)
        return

    # Common part for every exploded line, except 'summary'
    common = {k: v for k, v in block.items() if k in known_fields}

    # Optional keys that must appear only in the first exploded item
    first_only = block.pop("extras", None)

    for idx, part in enumerate(block["summary"]):
        new_block = dict(common)
        new_block["reasoning"] = part.get("text", "")
        if idx == 0 and first_only:
            new_block.update(first_only)
        if "index" in new_block:
            summary_index = part.get("index", 0)
            meaningful_idx = f"{new_block['index']}_{summary_index}"
            new_block["index"] = f"lc_rs_{meaningful_idx.encode().hex()}"

        yield cast("types.ReasoningContentBlock", new_block)


def _convert_to_v1_from_responses(message: AIMessage) -> list[types.ContentBlock]:
    """Convert a Responses message to v1 format."""

    def _iter_blocks() -> Iterator[types.ContentBlock]:
        for raw_block in message.content:
            if not isinstance(raw_block, dict):
                continue
            block = raw_block.copy()
            block_type = block.get("type")

            if block_type == "text":
                if "text" not in block:
                    block["text"] = ""
                if "annotations" in block:
                    block["annotations"] = [
                        _convert_annotation_to_v1(a) for a in block["annotations"]
                    ]
                if "index" in block:
                    block["index"] = f"lc_txt_{block['index']}"
                yield cast("types.TextContentBlock", block)

            elif block_type == "reasoning":
                yield from _explode_reasoning(block)

            elif block_type == "image_generation_call" and (
                result := block.get("result")
            ):
                new_block = {"type": "image", "base64": result}
                if output_format := block.get("output_format"):
                    new_block["mime_type"] = f"image/{output_format}"
                if "id" in block:
                    new_block["id"] = block["id"]
                if "index" in block:
                    new_block["index"] = f"lc_img_{block['index']}"
                for extra_key in (
                    "status",
                    "background",
                    "output_format",
                    "quality",
                    "revised_prompt",
                    "size",
                ):
                    if extra_key in block:
                        if "extras" not in new_block:
                            new_block["extras"] = {}
                        new_block["extras"][extra_key] = block[extra_key]
                yield cast("types.ImageContentBlock", new_block)

            elif block_type == "function_call":
                tool_call_block: (
                    types.ToolCall | types.InvalidToolCall | types.ToolCallChunk | None
                ) = None
                call_id = block.get("call_id", "")

                if (
                    isinstance(message, AIMessageChunk)
                    and len(message.tool_call_chunks) == 1
                    and message.chunk_position != "last"
                ):
                    tool_call_block = message.tool_call_chunks[0].copy()  # type: ignore[assignment]
                elif call_id:
                    for tool_call in message.tool_calls or []:
                        if tool_call.get("id") == call_id:
                            tool_call_block = {
                                "type": "tool_call",
                                "name": tool_call["name"],
                                "args": tool_call["args"],
                                "id": tool_call.get("id"),
                            }
                            break
                    else:
                        for invalid_tool_call in message.invalid_tool_calls or []:
                            if invalid_tool_call.get("id") == call_id:
                                tool_call_block = invalid_tool_call.copy()
                                break
                if tool_call_block:
                    if "id" in block:
                        if "extras" not in tool_call_block:
                            tool_call_block["extras"] = {}
                        tool_call_block["extras"]["item_id"] = block["id"]
                    if "index" in block:
                        tool_call_block["index"] = f"lc_tc_{block['index']}"
                    for extra_key in ("status", "namespace"):
                        if extra_key in block:
                            if "extras" not in tool_call_block:
                                tool_call_block["extras"] = {}
                            tool_call_block["extras"][extra_key] = block[extra_key]
                    yield tool_call_block

            elif block_type == "web_search_call":
                web_search_call = {
                    "type": "server_tool_call",
                    "name": "web_search",
                    "args": {},
                    "id": block["id"],
                }
                if "index" in block:
                    web_search_call["index"] = f"lc_wsc_{block['index']}"

                sources: dict[str, Any] | None = None
                if "action" in block and isinstance(block["action"], dict):
                    if "sources" in block["action"]:
                        sources = block["action"]["sources"]
                    web_search_call["args"] = {
                        k: v for k, v in block["action"].items() if k != "sources"
                    }
                for key in block:
                    if key not in {"type", "id", "action", "status", "index"}:
                        web_search_call[key] = block[key]

                yield cast("types.ServerToolCall", web_search_call)

                # If .content already has web_search_result, don't add
                if not any(
                    isinstance(other_block, dict)
                    and other_block.get("type") == "web_search_result"
                    and other_block.get("id") == block["id"]
                    for other_block in message.content
                ):
                    web_search_result = {
                        "type": "server_tool_result",
                        "tool_call_id": block["id"],
                    }
                    if sources:
                        web_search_result["output"] = {"sources": sources}

                    status = block.get("status")
                    if status == "failed":
                        web_search_result["status"] = "error"
                    elif status == "completed":
                        web_search_result["status"] = "success"
                    elif status:
                        web_search_result["extras"] = {"status": status}
                    if "index" in block and isinstance(block["index"], int):
                        web_search_result["index"] = f"lc_wsr_{block['index'] + 1}"
                    yield cast("types.ServerToolResult", web_search_result)

            elif block_type == "file_search_call":
                file_search_call = {
                    "type": "server_tool_call",
                    "name": "file_search",
                    "id": block["id"],
                    "args": {"queries": block.get("queries", [])},
                }
                if "index" in block:
                    file_search_call["index"] = f"lc_fsc_{block['index']}"

                for key in block:
                    if key not in {
                        "type",
                        "id",
                        "queries",
                        "results",
                        "status",
                        "index",
                    }:
                        file_search_call[key] = block[key]

                yield cast("types.ServerToolCall", file_search_call)

                file_search_result = {
                    "type": "server_tool_result",
                    "tool_call_id": block["id"],
                }
                if file_search_output := block.get("results"):
                    file_search_result["output"] = file_search_output

                status = block.get("status")
                if status == "failed":
                    file_search_result["status"] = "error"
                elif status == "completed":
                    file_search_result["status"] = "success"
                elif status:
                    file_search_result["extras"] = {"status": status}
                if "index" in block and isinstance(block["index"], int):
                    file_search_result["index"] = f"lc_fsr_{block['index'] + 1}"
                yield cast("types.ServerToolResult", file_search_result)

            elif block_type == "code_interpreter_call":
                code_interpreter_call = {
                    "type": "server_tool_call",
                    "name": "code_interpreter",
                    "id": block["id"],
                }
                if "code" in block:
                    code_interpreter_call["args"] = {"code": block["code"]}
                if "index" in block:
                    code_interpreter_call["index"] = f"lc_cic_{block['index']}"
                known_fields = {
                    "type",
                    "id",
                    "outputs",
                    "status",
                    "code",
                    "extras",
                    "index",
                }
                for key in block:
                    if key not in known_fields:
                        if "extras" not in code_interpreter_call:
                            code_interpreter_call["extras"] = {}
                        code_interpreter_call["extras"][key] = block[key]

                code_interpreter_result = {
                    "type": "server_tool_result",
                    "tool_call_id": block["id"],
                }
                if "outputs" in block:
                    code_interpreter_result["output"] = block["outputs"]

                status = block.get("status")
                if status == "failed":
                    code_interpreter_result["status"] = "error"
                elif status == "completed":
                    code_interpreter_result["status"] = "success"
                elif status:
                    code_interpreter_result["extras"] = {"status": status}
                if "index" in block and isinstance(block["index"], int):
                    code_interpreter_result["index"] = f"lc_cir_{block['index'] + 1}"

                yield cast("types.ServerToolCall", code_interpreter_call)
                yield cast("types.ServerToolResult", code_interpreter_result)

            elif block_type == "mcp_call":
                mcp_call = {
                    "type": "server_tool_call",
                    "name": "remote_mcp",
                    "id": block["id"],
                }
                if (arguments := block.get("arguments")) and isinstance(arguments, str):
                    try:
                        mcp_call["args"] = json.loads(block["arguments"])
                    except json.JSONDecodeError:
                        mcp_call["extras"] = {"arguments": arguments}
                if "name" in block:
                    if "extras" not in mcp_call:
                        mcp_call["extras"] = {}
                    mcp_call["extras"]["tool_name"] = block["name"]
                if "server_label" in block:
                    if "extras" not in mcp_call:
                        mcp_call["extras"] = {}
                    mcp_call["extras"]["server_label"] = block["server_label"]
                if "index" in block:
                    mcp_call["index"] = f"lc_mcp_{block['index']}"
                known_fields = {
                    "type",
                    "id",
                    "arguments",
                    "name",
                    "server_label",
                    "output",
                    "error",
                    "extras",
                    "index",
                }
                for key in block:
                    if key not in known_fields:
                        if "extras" not in mcp_call:
                            mcp_call["extras"] = {}
                        mcp_call["extras"][key] = block[key]

                yield cast("types.ServerToolCall", mcp_call)

                mcp_result = {
                    "type": "server_tool_result",
                    "tool_call_id": block["id"],
                }
                if mcp_output := block.get("output"):
                    mcp_result["output"] = mcp_output

                error = block.get("error")
                if error:
                    if "extras" not in mcp_result:
                        mcp_result["extras"] = {}
                    mcp_result["extras"]["error"] = error
                    mcp_result["status"] = "error"
                else:
                    mcp_result["status"] = "success"

                if "index" in block and isinstance(block["index"], int):
                    mcp_result["index"] = f"lc_mcpr_{block['index'] + 1}"
                yield cast("types.ServerToolResult", mcp_result)

            elif block_type == "mcp_list_tools":
                mcp_list_tools_call = {
                    "type": "server_tool_call",
                    "name": "mcp_list_tools",
                    "args": {},
                    "id": block["id"],
                }
                if "server_label" in block:
                    mcp_list_tools_call["extras"] = {}
                    mcp_list_tools_call["extras"]["server_label"] = block[
                        "server_label"
                    ]
                if "index" in block:
                    mcp_list_tools_call["index"] = f"lc_mlt_{block['index']}"
                known_fields = {
                    "type",
                    "id",
                    "name",
                    "server_label",
                    "tools",
                    "error",
                    "extras",
                    "index",
                }
                for key in block:
                    if key not in known_fields:
                        if "extras" not in mcp_list_tools_call:
                            mcp_list_tools_call["extras"] = {}
                        mcp_list_tools_call["extras"][key] = block[key]

                yield cast("types.ServerToolCall", mcp_list_tools_call)

                mcp_list_tools_result = {
                    "type": "server_tool_result",
                    "tool_call_id": block["id"],
                }
                if mcp_output := block.get("tools"):
                    mcp_list_tools_result["output"] = mcp_output

                error = block.get("error")
                if error:
                    if "extras" not in mcp_list_tools_result:
                        mcp_list_tools_result["extras"] = {}
                    mcp_list_tools_result["extras"]["error"] = error
                    mcp_list_tools_result["status"] = "error"
                else:
                    mcp_list_tools_result["status"] = "success"

                if "index" in block and isinstance(block["index"], int):
                    mcp_list_tools_result["index"] = f"lc_mltr_{block['index'] + 1}"
                yield cast("types.ServerToolResult", mcp_list_tools_result)

            elif (
                block_type == "tool_search_call" and block.get("execution") == "server"
            ):
                tool_search_call: dict[str, Any] = {
                    "type": "server_tool_call",
                    "name": "tool_search",
                    "id": block["id"],
                    "args": block.get("arguments", {}),
                }
                if "index" in block:
                    tool_search_call["index"] = f"lc_tsc_{block['index']}"
                extras: dict[str, Any] = {}
                known = {"type", "id", "arguments", "index"}
                for key in block:
                    if key not in known:
                        extras[key] = block[key]
                if extras:
                    tool_search_call["extras"] = extras
                yield cast("types.ServerToolCall", tool_search_call)

            elif (
                block_type == "tool_search_output"
                and block.get("execution") == "server"
            ):
                tool_search_output: dict[str, Any] = {
                    "type": "server_tool_result",
                    "tool_call_id": block["id"],
                    "output": {"tools": block.get("tools", [])},
                }
                status = block.get("status")
                if status == "failed":
                    tool_search_output["status"] = "error"
                elif status == "completed":
                    tool_search_output["status"] = "success"
                if "index" in block and isinstance(block["index"], int):
                    tool_search_output["index"] = f"lc_tso_{block['index']}"
                extras_out: dict[str, Any] = {"name": "tool_search"}
                known_out = {"type", "id", "status", "tools", "index"}
                for key in block:
                    if key not in known_out:
                        extras_out[key] = block[key]
                if extras_out:
                    tool_search_output["extras"] = extras_out
                yield cast("types.ServerToolResult", tool_search_output)

            elif block_type in types.KNOWN_BLOCK_TYPES:
                yield cast("types.ContentBlock", block)
            else:
                new_block = {"type": "non_standard", "value": block}
                if "index" in new_block["value"]:
                    new_block["index"] = f"lc_ns_{new_block['value'].pop('index')}"
                yield cast("types.NonStandardContentBlock", new_block)

    return list(_iter_blocks())


def translate_content(message: AIMessage) -> list[types.ContentBlock]:
    """Derive standard content blocks from a message with OpenAI content.

    Args:
        message: The message to translate.

    Returns:
        The derived content blocks.
    """
    if isinstance(message.content, str):
        return _convert_to_v1_from_chat_completions(message)
    message = _convert_from_v03_ai_message(message)
    return _convert_to_v1_from_responses(message)


def translate_content_chunk(message: AIMessageChunk) -> list[types.ContentBlock]:
    """Derive standard content blocks from a message chunk with OpenAI content.

    Args:
        message: The message chunk to translate.

    Returns:
        The derived content blocks.
    """
    if isinstance(message.content, str):
        return _convert_to_v1_from_chat_completions_chunk(message)
    message = _convert_from_v03_ai_message(message)  # type: ignore[assignment]
    return _convert_to_v1_from_responses(message)


def _register_openai_translator() -> None:
    """Register the OpenAI translator with the central registry.

    Run automatically when the module is imported.
    """
    from langchain_core.messages.block_translators import (  # noqa: PLC0415
        register_translator,
    )

    register_translator("openai", translate_content, translate_content_chunk)


_register_openai_translator()
