# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0

import json
import os
from dataclasses import dataclass
from typing import Any
from unittest.mock import patch

import pytest
from openai.types.chat import ChatCompletion, ChatCompletionMessage
from openai.types.chat.chat_completion import Choice

from haystack import Pipeline, SuperComponent, component
from haystack.components.agents import Agent, State
from haystack.components.builders import PromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.core.pipeline.utils import _deepcopy_with_exceptions
from haystack.dataclasses import ChatMessage, ChatRole, Document
from haystack.tools import ComponentTool, ToolsType
from test.tools.test_parameters_schema_utils import BYTE_STREAM_SCHEMA, DOCUMENT_SCHEMA, SPARSE_EMBEDDING_SCHEMA

# Component and Model Definitions


@component
class SimpleComponentUsingChatMessages:
    """A simple component that generates text."""

    @component.output_types(reply=str)
    def run(self, messages: list[ChatMessage]) -> dict[str, str]:
        """
        A simple component that generates text.

        :param messages: Users messages
        :return: A dictionary with the generated text.
        """
        return {"reply": f"Hello, {messages[0].text}!"}


@component
class SimpleComponent:
    """A simple component that generates text."""

    def warm_up(self):
        """
        Prepare the component for use.
        """

    @component.output_types(reply=str)
    def run(self, text: str) -> dict[str, str]:
        """
        A simple component that generates text.

        :param text: user's name
        :return: A dictionary with the generated text.
        """
        return {"reply": f"Hello, {text}!"}


def reply_formatter(input_text: str) -> str:
    return f"Formatted reply: {input_text}"


@dataclass
class User:
    """A simple user dataclass."""

    name: str = "Anonymous"
    age: int = 0


@component
class UserGreeter:
    """A simple component that processes a User."""

    @component.output_types(message=str)
    def run(self, user: User) -> dict[str, str]:
        """
        A simple component that processes a User.

        :param user: The User object to process.
        :return: A dictionary with a message about the user.
        """
        return {"message": f"User {user.name} is {user.age} years old"}


@component
class ListProcessor:
    """A component that processes a list of strings."""

    @component.output_types(concatenated=str)
    def run(self, texts: list[str]) -> dict[str, str]:
        """
        Concatenates a list of strings into a single string.

        :param texts: The list of strings to concatenate.
        :return: A dictionary with the concatenated string.
        """
        return {"concatenated": " ".join(texts)}


@dataclass
class Address:
    """A dataclass representing a physical address."""

    street: str
    city: str


@dataclass
class Person:
    """A person with an address."""

    name: str
    address: Address


@component
class PersonProcessor:
    """A component that processes a Person with nested Address."""

    @component.output_types(info=str)
    def run(self, person: Person) -> dict[str, str]:
        """
        Creates information about the person.

        :param person: The Person to process.
        :return: A dictionary with the person's information.
        """
        return {"info": f"{person.name} lives at {person.address.street}, {person.address.city}."}


@component
class DocumentProcessor:
    """A component that processes a list of Documents."""

    @component.output_types(concatenated=str)
    def run(self, documents: list[Document], top_k: int = 5) -> dict[str, str]:
        """
        Concatenates the content of multiple documents with newlines.

        :param documents: List of Documents whose content will be concatenated
        :param top_k: The number of top documents to concatenate
        :returns: Dictionary containing the concatenated document contents
        """
        return {"concatenated": "\n".join(doc.content for doc in documents[:top_k] if doc.content)}


@component
class FakeChatGenerator:
    def __init__(self, messages: list[ChatMessage]):
        self.messages = messages

    @component.output_types(replies=list[ChatMessage])
    def run(
        self,
        messages: list[ChatMessage],
        generation_kwargs: dict[str, Any] | None = None,
        *,
        tools: ToolsType | None = None,
    ) -> dict[str, list[ChatMessage]]:
        return {"replies": self.messages}


def output_handler(old, new):
    """
    Output handler to test serialization.
    """
    return old + new


class TestComponentTool:
    def test_from_component_basic(self):
        tool = ComponentTool(component=SimpleComponent())

        assert tool.name == "simple_component"
        assert tool.description == "A simple component that generates text."
        assert tool.parameters == {
            "type": "object",
            "properties": {"text": {"type": "string", "description": "user's name"}},
            "required": ["text"],
        }

        # Test tool invocation
        result = tool.invoke(text="world")
        assert isinstance(result, dict)
        assert "reply" in result
        assert result["reply"] == "Hello, world!"

    def test_from_component_long_description(self):
        tool = ComponentTool(component=SimpleComponent(), description="".join(["A"] * 1024))
        assert len(tool.description) == 1024

    def test_from_component_with_inputs_from_state(self):
        tool = ComponentTool(component=SimpleComponent(), inputs_from_state={"text": "text"})
        assert tool.inputs_from_state == {"text": "text"}
        # Inputs should be excluded from schema generation
        assert tool.parameters == {"type": "object", "properties": {}}

    def test_from_component_with_inputs_from_state_different_names(self):
        tool = ComponentTool(component=SimpleComponent(), inputs_from_state={"state_text": "text"})
        assert tool.inputs_from_state == {"state_text": "text"}
        # Inputs should be excluded from schema generation
        assert tool.parameters == {"type": "object", "properties": {}}

    def test_from_component_with_invalid_inputs_from_state_nested_dict(self):
        """Test that ComponentTool rejects nested dict format for inputs_from_state"""
        with pytest.raises(TypeError, match="must be str, not dict"):
            ComponentTool(component=SimpleComponent(), inputs_from_state={"documents": {"source": "documents"}})  # type: ignore[dict-item]

    def test_from_component_with_outputs_to_state(self):
        tool = ComponentTool(component=SimpleComponent(), outputs_to_state={"replies": {"source": "reply"}})
        assert tool.outputs_to_state == {"replies": {"source": "reply"}}

    def test_from_component_with_invalid_outputs_to_state_source(self):
        """Test that ComponentTool validates outputs_to_state source against component outputs"""
        with pytest.raises(ValueError, match="unknown output"):
            ComponentTool(component=SimpleComponent(), outputs_to_state={"result": {"source": "nonexistent"}})

    def test_from_component_with_dataclass(self):
        tool = ComponentTool(component=UserGreeter())
        assert tool.parameters == {
            "$defs": {
                "User": {
                    "properties": {
                        "name": {"description": "Field 'name' of 'User'.", "type": "string", "default": "Anonymous"},
                        "age": {"description": "Field 'age' of 'User'.", "type": "integer", "default": 0},
                    },
                    "type": "object",
                }
            },
            "properties": {"user": {"$ref": "#/$defs/User", "description": "The User object to process."}},
            "required": ["user"],
            "type": "object",
        }

        assert tool.name == "user_greeter"
        assert tool.description == "A simple component that processes a User."

        # Test tool invocation
        result = tool.invoke(user={"name": "Alice", "age": 30})
        assert isinstance(result, dict)
        assert "message" in result
        assert result["message"] == "User Alice is 30 years old"

    def test_from_component_with_list_input(self):
        tool = ComponentTool(
            component=ListProcessor(), name="list_processing_tool", description="A tool that concatenates strings"
        )

        assert tool.parameters == {
            "type": "object",
            "properties": {
                "texts": {
                    "type": "array",
                    "description": "The list of strings to concatenate.",
                    "items": {"type": "string"},
                }
            },
            "required": ["texts"],
        }

        # Test tool invocation
        result = tool.invoke(texts=["hello", "world"])
        assert isinstance(result, dict)
        assert "concatenated" in result
        assert result["concatenated"] == "hello world"

    def test_from_component_with_nested_dataclass(self):
        tool = ComponentTool(
            component=PersonProcessor(), name="person_tool", description="A tool that processes people"
        )

        assert tool.parameters == {
            "$defs": {
                "Address": {
                    "properties": {
                        "street": {"description": "Field 'street' of 'Address'.", "type": "string"},
                        "city": {"description": "Field 'city' of 'Address'.", "type": "string"},
                    },
                    "required": ["street", "city"],
                    "type": "object",
                },
                "Person": {
                    "properties": {
                        "name": {"description": "Field 'name' of 'Person'.", "type": "string"},
                        "address": {"$ref": "#/$defs/Address", "description": "Field 'address' of 'Person'."},
                    },
                    "required": ["name", "address"],
                    "type": "object",
                },
            },
            "properties": {"person": {"$ref": "#/$defs/Person", "description": "The Person to process."}},
            "required": ["person"],
            "type": "object",
        }

        # Test tool invocation
        result = tool.invoke(person={"name": "Diana", "address": {"street": "123 Elm Street", "city": "Metropolis"}})
        assert isinstance(result, dict)
        assert "info" in result
        assert result["info"] == "Diana lives at 123 Elm Street, Metropolis."

    def test_from_component_with_list_of_documents(self):
        tool = ComponentTool(
            component=DocumentProcessor(),
            name="document_processor",
            description="A tool that concatenates document contents",
        )

        assert tool.parameters == {
            "$defs": {
                "ByteStream": BYTE_STREAM_SCHEMA,
                "Document": DOCUMENT_SCHEMA,
                "SparseEmbedding": SPARSE_EMBEDDING_SCHEMA,
            },
            "properties": {
                "documents": {
                    "description": "List of Documents whose content will be concatenated",
                    "items": {"$ref": "#/$defs/Document"},
                    "type": "array",
                },
                "top_k": {"description": "The number of top documents to concatenate", "type": "integer", "default": 5},
            },
            "required": ["documents"],
            "type": "object",
        }

        # Test tool invocation
        result = tool.invoke(documents=[{"content": "First document"}, {"content": "Second document"}])
        assert isinstance(result, dict)
        assert "concatenated" in result
        assert result["concatenated"] == "First document\nSecond document"

    def test_from_component_with_dynamic_input_types(self):
        builder = PromptBuilder(template="Hello, {{name}}!")
        tool = ComponentTool(component=builder, name="prompt_builder_tool")
        assert tool.parameters == {
            "properties": {
                "name": {"description": "Input 'name' for the component."},
                "template": {
                    "anyOf": [{"type": "string"}, {"type": "null"}],
                    "default": None,
                    "description": "An optional string template to overwrite PromptBuilder's default template. If "
                    "None, the default template\nprovided at initialization is used.",
                },
                "template_variables": {
                    "anyOf": [{"additionalProperties": True, "type": "object"}, {"type": "null"}],
                    "default": None,
                    "description": "An optional dictionary of template variables to overwrite the pipeline variables.",
                },
            },
            "required": ["name"],
            "type": "object",
        }

    def test_from_component_with_invalid_component(self):
        class NotAComponent:
            def foo(self, text: str) -> dict[str, str]:
                return {"reply": f"Hello, {text}!"}

        not_a_component = NotAComponent()

        with pytest.raises(TypeError):
            ComponentTool(component=not_a_component, name="invalid_tool", description="This should fail")  # type: ignore[arg-type]

    def test_component_invoker_with_chat_message_input(self):
        tool = ComponentTool(
            component=SimpleComponentUsingChatMessages(), name="simple_tool", description="A simple tool"
        )
        result = tool.invoke(messages=[ChatMessage.from_user(text="world")])
        assert result == {"reply": "Hello, world!"}

    def test_component_tool_with_super_component_docstrings(self, monkeypatch):
        """Test that ComponentTool preserves docstrings from underlying pipeline components in SuperComponents."""

        @component
        class AnnotatedComponent:
            """An annotated component with descriptive parameter docstrings."""

            @component.output_types(result=str)
            def run(self, text: str, number: int = 42) -> dict[str, str]:
                """
                Process inputs and return result.

                :param text: A detailed description of the text parameter that should be preserved
                :param number: A detailed description of the number parameter that should be preserved
                """
                return {"result": f"Processed: {text} and {number}"}

        # Create a pipeline with the annotated component
        pipeline = Pipeline()
        pipeline.add_component("processor", AnnotatedComponent())
        # Create SuperComponent with mapping
        super_comp = SuperComponent(
            pipeline=pipeline,
            input_mapping={"input_text": ["processor.text"], "input_number": ["processor.number"]},
            output_mapping={"processor.result": "processed_result"},
        )

        # Create ComponentTool from SuperComponent
        tool = ComponentTool(component=super_comp, name="text_processor")

        # Verify that schema includes the per-parameter docstrings from the original component
        assert tool.parameters == {
            "type": "object",
            "properties": {
                "input_text": {
                    "type": "string",
                    "description": "Provided to the 'processor' component as: 'A detailed description of the text "
                    "parameter that should be preserved'.",
                },
                "input_number": {
                    "type": "integer",
                    "description": "Provided to the 'processor' component as: 'A detailed description of the number "
                    "parameter that should be preserved'.",
                },
            },
            "required": ["input_text"],
        }

        # Test the tool functionality works
        result = tool.invoke(input_text="Hello", input_number=42)
        assert result["processed_result"] == "Processed: Hello and 42"

    def test_component_tool_with_multiple_mapped_docstrings(self):
        """
        Test ComponentTool combines docstrings from multiple components when a single input maps to multiple components.
        """

        @component
        class ComponentA:
            """Component A with descriptive docstrings."""

            @component.output_types(output_a=str)
            def run(self, query: str) -> dict[str, str]:
                """
                Process query in component A.

                :param query: The query string for component A
                """
                return {"output_a": f"A processed: {query}"}

        @component
        class ComponentB:
            """Component B with descriptive docstrings."""

            @component.output_types(output_b=str)
            def run(self, text: str) -> dict[str, str]:
                """
                Process text in component B.

                :param text: Text to process in component B
                """
                return {"output_b": f"B processed: {text}"}

        # Create a pipeline with both components
        pipeline = Pipeline()
        pipeline.add_component("comp_a", ComponentA())
        pipeline.add_component("comp_b", ComponentB())

        # Create SuperComponent with a single input mapped to both components
        super_comp = SuperComponent(
            pipeline=pipeline, input_mapping={"combined_input": ["comp_a.query", "comp_b.text"]}
        )

        # Create ComponentTool from SuperComponent
        tool = ComponentTool(component=super_comp, name="combined_processor")

        # Verify that schema includes combined per-parameter docstrings from both components
        assert tool.parameters == {
            "type": "object",
            "properties": {
                "combined_input": {
                    "type": "string",
                    "description": "Provided to the 'comp_a' component as: 'The query string for component A', and "
                    "Provided to the 'comp_b' component as: 'Text to process in component B'.",
                }
            },
            "required": ["combined_input"],
        }

        # Test the tool functionality works
        result = tool.invoke(combined_input="test input")
        assert result["output_a"] == "A processed: test input"
        assert result["output_b"] == "B processed: test input"

    def test_warm_up_is_idempotent(self):
        """Test that calling warm_up multiple times only warms up the component once."""
        from unittest.mock import MagicMock, patch

        component = SimpleComponent()

        tool = ComponentTool(component=component)

        with patch.object(component, "warm_up", MagicMock()) as mock_warm_up:
            # Call warm_up multiple times
            tool.warm_up()
            tool.warm_up()
            tool.warm_up()

            # Component's warm_up should only be called once
            mock_warm_up.assert_called_once()

    def test_from_component_with_callable_params_skipped(self, monkeypatch):
        monkeypatch.setenv("OPENAI_API_KEY", "test-api-key")
        agent = Agent(chat_generator=OpenAIChatGenerator(model="gpt-4o-mini"))

        tool = ComponentTool(
            component=agent,
            name="agent_tool",
            description="An agent tool",
            outputs_to_string={"source": "last_message"},
        )

        assert tool.name == "agent_tool"
        assert tool.description == "An agent tool"

        param_names = list(tool.parameters.get("properties", {}).keys())
        assert "snapshot_callback" not in param_names
        assert "streaming_callback" not in param_names
        assert "messages" in param_names

    def test_from_component_with_state_param_excluded_from_schema(self):
        @component
        class ComponentWithState:
            """A component that takes State as a direct input."""

            @component.output_types(result=str)
            def run(self, query: str, state: State) -> dict:
                return {"result": query}

        tool = ComponentTool(component=ComponentWithState(), name="state_comp", description="test")

        param_names = list(tool.parameters.get("properties", {}).keys())
        assert "state" not in param_names
        assert "query" in param_names

    def test_from_component_with_optional_state_param_excluded_from_schema(self):
        @component
        class ComponentWithOptionalState:
            """A component that takes Optional[State] as an input (e.g. Agent tool-calling style)."""

            @component.output_types(result=str)
            def run(self, query: str, state: State | None = None) -> dict:
                return {"result": query}

        tool = ComponentTool(component=ComponentWithOptionalState(), name="opt_state_comp", description="test")

        param_names = list(tool.parameters.get("properties", {}).keys())
        assert "state" not in param_names
        assert "query" in param_names

    def test_component_invoker_with_agent(self):
        """Tests that Agent as a ComponentTool can be invoked when calling it with a list of dicts"""
        agent = Agent(chat_generator=FakeChatGenerator(messages=[ChatMessage.from_assistant("Answer")]))
        tool = ComponentTool(
            component=agent,
            name="agent_tool",
            description="An agent tool",
            outputs_to_string={"source": "last_message"},
        )
        result = tool.invoke(messages=[{"role": "user", "content": [{"text": "A 4-day trip in the south of France"}]}])
        assert result["last_message"] == ChatMessage.from_assistant("Answer")

    def test_convert_param_union_with_list_arm(self):
        @component
        class ComponentWithUnionMessages:
            @component.output_types(reply=str)
            def run(self, messages: list[ChatMessage] | str) -> dict:
                return {"reply": "ok"}

        tool = ComponentTool(component=ComponentWithUnionMessages())
        result = tool._convert_param([{"role": "user", "content": "Hello"}], list[ChatMessage] | str)  # type: ignore[arg-type]
        assert result == [ChatMessage.from_user("Hello")]


def _agent_tool_messages(result: dict[str, Any]) -> list[ChatMessage]:
    return [message for message in result["agent"]["messages"] if message.is_from(ChatRole.TOOL)]


class TestComponentToolInAgent:
    @pytest.mark.skipif(not os.environ.get("OPENAI_API_KEY"), reason="OPENAI_API_KEY not set")
    @pytest.mark.integration
    def test_component_tool(self):
        # Create component and convert it to tool
        tool = ComponentTool(
            component=SimpleComponent(),
            name="hello_tool",
            description="A tool that generates a greeting message for the user",
        )

        pipeline = Pipeline()
        pipeline.add_component("agent", Agent(chat_generator=OpenAIChatGenerator(), tools=[tool]))

        message = ChatMessage.from_user(text="Using a single tool call, greet Vladimir")

        # Run pipeline
        result = pipeline.run({"agent": {"messages": [message]}})

        # Check results
        tool_messages = _agent_tool_messages(result)
        assert len(tool_messages) == 1

        tool_message = tool_messages[0]
        assert tool_message.is_from(ChatRole.TOOL)
        tool_call_result = tool_message.tool_call_result
        assert tool_call_result is not None
        assert "Vladimir" in tool_call_result.result
        assert not tool_call_result.error

    @pytest.mark.skipif(not os.environ.get("OPENAI_API_KEY"), reason="OPENAI_API_KEY not set")
    @pytest.mark.integration
    @pytest.mark.flaky(reruns=3, reruns_delay=10)
    def test_component_tool_openai_tools_strict(self):
        # Create component and convert it to tool
        tool = ComponentTool(
            component=SimpleComponent(),
            name="hello_tool",
            description="A tool that generates a greeting message for the user",
        )

        pipeline = Pipeline()
        pipeline.add_component("agent", Agent(chat_generator=OpenAIChatGenerator(tools_strict=True), tools=[tool]))

        message = ChatMessage.from_user(text="Using tools, greet Vladimir")

        # Run pipeline
        result = pipeline.run({"agent": {"messages": [message]}})

        # Check results
        tool_messages = _agent_tool_messages(result)
        assert len(tool_messages) == 1

        tool_message = tool_messages[0]
        assert tool_message.is_from(ChatRole.TOOL)
        tool_call_result = tool_message.tool_call_result
        assert tool_call_result is not None
        assert "Vladimir" in tool_call_result.result
        assert not tool_call_result.error

    @pytest.mark.skipif(not os.environ.get("OPENAI_API_KEY"), reason="OPENAI_API_KEY not set")
    @pytest.mark.integration
    def test_user_greeter(self):
        tool = ComponentTool(
            component=UserGreeter(), name="user_greeter", description="A tool that greets users with their name and age"
        )

        pipeline = Pipeline()
        pipeline.add_component("agent", Agent(chat_generator=OpenAIChatGenerator(), tools=[tool]))

        message = ChatMessage.from_user(text="Greet the user Alice who is 30 years old")

        result = pipeline.run({"agent": {"messages": [message]}})
        tool_messages = _agent_tool_messages(result)
        assert len(tool_messages) == 1

        tool_message = tool_messages[0]
        assert tool_message.is_from(ChatRole.TOOL)
        tool_call_result = tool_message.tool_call_result
        assert tool_call_result is not None
        assert tool_call_result.result == json.dumps({"message": "User Alice is 30 years old"})
        assert not tool_call_result.error

    @pytest.mark.skipif(not os.environ.get("OPENAI_API_KEY"), reason="OPENAI_API_KEY not set")
    @pytest.mark.integration
    def test_list_processor(self):
        tool = ComponentTool(
            component=ListProcessor(), name="list_processor", description="A tool that concatenates a list of strings"
        )

        pipeline = Pipeline()
        pipeline.add_component("agent", Agent(chat_generator=OpenAIChatGenerator(), tools=[tool]))

        # Be explicit about using tools, otherwise model will ignore the tool call and return the result directly.
        message = ChatMessage.from_user(text="Using tools, join these words: hello, beautiful, world")

        result = pipeline.run({"agent": {"messages": [message]}})
        tool_messages = _agent_tool_messages(result)
        # The model may issue one or more parallel tool calls, so check the content across all of them.
        assert len(tool_messages) >= 1

        combined = ""
        for tool_message in tool_messages:
            assert tool_message.is_from(ChatRole.TOOL)
            tool_call_result = tool_message.tool_call_result
            assert tool_call_result is not None and not tool_call_result.error
            assert isinstance(tool_call_result.result, str)
            assert "concatenated" in tool_call_result.result
            combined += " " + tool_call_result.result
        # Normalize whitespace and check the concatenated output contains the expected words.
        normalized_result = " ".join(combined.split())
        assert "hello" in normalized_result
        assert "beautiful" in normalized_result
        assert "world" in normalized_result

    @pytest.mark.skipif(not os.environ.get("OPENAI_API_KEY"), reason="OPENAI_API_KEY not set")
    @pytest.mark.integration
    def test_person_processor(self):
        tool = ComponentTool(
            component=PersonProcessor(),
            name="person_processor",
            description="A tool that processes information about a person and their address",
        )

        pipeline = Pipeline()
        pipeline.add_component("agent", Agent(chat_generator=OpenAIChatGenerator(), tools=[tool]))

        message = ChatMessage.from_user(
            text="Process information about the person Diana who lives at 123 Elm Street in Metropolis"
        )

        result = pipeline.run({"agent": {"messages": [message]}})
        tool_messages = _agent_tool_messages(result)
        assert len(tool_messages) == 1

        tool_message = tool_messages[0]
        assert tool_message.is_from(ChatRole.TOOL)
        tool_call_result = tool_message.tool_call_result
        assert tool_call_result is not None
        assert "Diana" in tool_call_result.result and "Metropolis" in tool_call_result.result
        assert not tool_call_result.error

    @pytest.mark.skipif(not os.environ.get("OPENAI_API_KEY"), reason="OPENAI_API_KEY not set")
    @pytest.mark.integration
    def test_document_processor(self):
        tool = ComponentTool(
            component=DocumentProcessor(),
            name="document_processor",
            description="A tool that concatenates the content of multiple documents",
        )

        pipeline = Pipeline()
        pipeline.add_component("agent", Agent(chat_generator=OpenAIChatGenerator(), tools=[tool]))

        message = ChatMessage.from_user(
            text="Concatenate these documents: First one says 'Hello world' and second one says 'Goodbye world' and "
            "third one says 'Hello again', but use top_k=2. Set only content field of the document only. Do "
            "not set id, meta, score, embedding, sparse_embedding, dataframe, blob fields."
        )

        result = pipeline.run({"agent": {"messages": [message]}})

        tool_messages = _agent_tool_messages(result)
        assert len(tool_messages) == 1

        tool_message = tool_messages[0]
        assert tool_message.is_from(ChatRole.TOOL)
        tool_call_result = tool_message.tool_call_result
        assert tool_call_result is not None
        assert isinstance(tool_call_result.result, str)
        result = json.loads(tool_call_result.result)
        assert "concatenated" in result
        assert "Hello world" in result["concatenated"]
        assert "Goodbye world" in result["concatenated"]
        assert not tool_call_result.error

    @pytest.mark.skipif(not os.environ.get("OPENAI_API_KEY"), reason="OPENAI_API_KEY not set")
    @pytest.mark.integration
    def test_lost_in_middle_ranker(self):
        from haystack.components.rankers import LostInTheMiddleRanker

        tool = ComponentTool(
            component=LostInTheMiddleRanker(),
            name="lost_in_middle_ranker",
            description="A tool that ranks documents using the Lost in the Middle algorithm and returns top k results",
        )

        pipeline = Pipeline()
        pipeline.add_component("agent", Agent(chat_generator=OpenAIChatGenerator(), tools=[tool]))

        message = ChatMessage.from_user(
            text="I have three documents with content: 'First doc', 'Middle doc', and 'Last doc'. Rank them top_k=2. "
            "Set only content field of the document only. Do not set id, meta, score, embedding, "
            "sparse_embedding, dataframe, blob fields."
        )

        result = pipeline.run({"agent": {"messages": [message]}})

        tool_messages = _agent_tool_messages(result)
        assert len(tool_messages) == 1
        tool_message = tool_messages[0]
        assert tool_message.is_from(ChatRole.TOOL)

    def test_serde(self, monkeypatch):
        monkeypatch.setenv("OPENAI_API_KEY", "test-key")

        # Create the component and tool
        tool = ComponentTool(component=SimpleComponent(), name="hello_tool", description="A simple greeting tool")

        pipeline = Pipeline()
        pipeline.add_component("agent", Agent(chat_generator=OpenAIChatGenerator(), tools=[tool]))

        # Serialize to dict and verify structure
        pipeline_dict = pipeline.to_dict()
        assert pipeline_dict["components"]["agent"]["type"] == "haystack.components.agents.agent.Agent"
        assert len(pipeline_dict["components"]["agent"]["init_parameters"]["tools"]) == 1

        tool_dict = pipeline_dict["components"]["agent"]["init_parameters"]["tools"][0]
        assert tool_dict["type"] == "haystack.tools.component_tool.ComponentTool"
        assert tool_dict["data"]["name"] == "hello_tool"
        assert tool_dict["data"]["component"]["type"] == "test_component_tool.SimpleComponent"

        # Test round-trip serialization
        pipeline_yaml = pipeline.dumps()
        new_pipeline = Pipeline.loads(pipeline_yaml)
        assert new_pipeline.to_dict() == pipeline_dict

    def test_component_tool_serde(self):
        tool = ComponentTool(
            component=SimpleComponent(),
            name="simple_tool",
            description="A simple tool",
            outputs_to_string={"source": "reply", "handler": reply_formatter},
            inputs_from_state={"test": "text"},
            outputs_to_state={"output": {"source": "reply", "handler": output_handler}},
        )

        # Test serialization
        expected_tool_dict = {
            "type": "haystack.tools.component_tool.ComponentTool",
            "data": {
                "component": {"type": "test_component_tool.SimpleComponent", "init_parameters": {}},
                "name": "simple_tool",
                "description": "A simple tool",
                "parameters": None,
                "outputs_to_string": {"source": "reply", "handler": "test_component_tool.reply_formatter"},
                "inputs_from_state": {"test": "text"},
                "outputs_to_state": {"output": {"source": "reply", "handler": "test_component_tool.output_handler"}},
            },
        }
        tool_dict = tool.to_dict()
        assert tool_dict == expected_tool_dict

        # Test deserialization
        new_tool = ComponentTool.from_dict(expected_tool_dict)
        assert new_tool.name == tool.name
        assert new_tool.description == tool.description
        assert new_tool.parameters == tool.parameters
        assert new_tool.outputs_to_string == tool.outputs_to_string
        assert new_tool.inputs_from_state == tool.inputs_from_state
        assert new_tool.outputs_to_state == tool.outputs_to_state
        assert isinstance(new_tool._component, SimpleComponent)

    def test_pipeline_component_fails(self):
        comp = SimpleComponent()

        # Create a pipeline and add the component to it
        pipeline = Pipeline()
        pipeline.add_component("simple", comp)

        # Try to create a tool from the component and it should fail because the component has been added to a pipeline
        # and thus can't be used as tool
        with pytest.raises(ValueError, match="Component has been added to a pipeline"):
            ComponentTool(component=comp)

    def test_deepcopy_with_jinja_based_component(self):
        builder = PromptBuilder("{{query}}")
        tool = ComponentTool(component=builder)
        assert tool.function is not None
        result = tool.function(query="Hello")
        tool_copy = _deepcopy_with_exceptions(tool)
        assert tool_copy.function is not None
        result_from_copy = tool_copy.function(query="Hello")
        assert "prompt" in result_from_copy
        assert result_from_copy["prompt"] == result["prompt"]

    def test_jinja_based_component_tool(self, monkeypatch):
        monkeypatch.setenv("OPENAI_API_KEY", "test-key")

        with patch("openai.resources.chat.completions.Completions.create") as mock_create:
            mock_create.return_value = ChatCompletion(
                id="test",
                model="gpt-4o-mini",
                object="chat.completion",
                choices=[
                    Choice(
                        finish_reason="length",
                        index=0,
                        message=ChatCompletionMessage(role="assistant", content="A response from the model"),
                    )
                ],
                created=1234567890,
            )

            tool = ComponentTool(component=PromptBuilder("{{query}}"))
            pipeline = Pipeline()
            pipeline.add_component("llm", OpenAIChatGenerator())
            result = pipeline.run({"llm": {"messages": [ChatMessage.from_user(text="Hello")], "tools": [tool]}})

        assert result["llm"]["replies"][0].text == "A response from the model"


@component
class SyncOnlyComponent:
    @component.output_types(reply=str)
    def run(self, text: str) -> dict[str, str]:
        return {"reply": f"sync:{text}"}


@component
class DualModeComponent:
    @component.output_types(reply=str)
    def run(self, text: str) -> dict[str, str]:
        return {"reply": f"sync:{text}"}

    @component.output_types(reply=str)
    async def run_async(self, text: str) -> dict[str, str]:
        return {"reply": f"async:{text}"}


@pytest.fixture
def sync_tool():
    return ComponentTool(component=SyncOnlyComponent())


@pytest.fixture
def dual_tool():
    return ComponentTool(component=DualModeComponent())


class TestComponentToolAsync:
    def test_async_function_is_wired_only_when_component_has_run_async(self, sync_tool, dual_tool):
        assert sync_tool.function is not None
        assert sync_tool.async_function is None
        assert dual_tool.function is not None
        assert dual_tool.async_function is not None

    @pytest.mark.asyncio
    async def test_invoke_async_uses_run_async_when_available(self, dual_tool):
        assert await dual_tool.invoke_async(text="hi") == {"reply": "async:hi"}

    @pytest.mark.asyncio
    async def test_invoke_async_falls_back_to_run_for_sync_only_component(self, sync_tool):
        assert await sync_tool.invoke_async(text="hi") == {"reply": "sync:hi"}

    def test_invoke_uses_run_on_dual_mode_component(self, dual_tool):
        assert dual_tool.invoke(text="hi") == {"reply": "sync:hi"}