# Copyright (c) Microsoft. All rights reserved.

import asyncio
from collections.abc import Awaitable, Callable
from typing import Any

import pytest

from agent_framework import (
    Agent,
    ChatResponse,
    ChatResponseUpdate,
    Content,
    Message,
    SupportsChatGetResponse,
    chat_middleware,
    tool,
)
from agent_framework._compaction import (
    EXCLUDED_KEY,
    GROUP_ANNOTATION_KEY,
    GROUP_ID_KEY,
    CharacterEstimatorTokenizer,
    SlidingWindowStrategy,
    TokenBudgetComposedStrategy,
    annotate_message_groups,
    included_token_count,
)
from agent_framework._middleware import FunctionInvocationContext, FunctionMiddleware, MiddlewareTermination


def _group_id(message: Message) -> str | None:
    annotation = message.additional_properties.get(GROUP_ANNOTATION_KEY)
    if not isinstance(annotation, dict):
        return None
    value = annotation.get(GROUP_ID_KEY)
    return value if isinstance(value, str) else None


def _build_approved_tool_roundtrip(
    *,
    call_id: str,
    approval_id: str,
    tool_name: str,
) -> tuple[Content, Content, Content]:
    function_call = Content.from_function_call(call_id=call_id, name=tool_name, arguments="{}")
    approval_request = Content.from_function_approval_request(id=approval_id, function_call=function_call)
    approval_response = approval_request.to_function_approval_response(approved=True)
    return function_call, approval_request, approval_response


async def test_base_client_with_function_calling(chat_client_base: SupportsChatGetResponse):
    exec_counter = 0

    @tool(name="test_function", approval_mode="never_require")
    def ai_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Processed {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="test_function", arguments='{"arg1": "value1"}')
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]
    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [ai_func]}
    )
    assert exec_counter == 1
    assert len(response.messages) == 3
    assert response.messages[0].role == "assistant"
    assert response.messages[0].contents[0].type == "function_call"
    assert response.messages[0].contents[0].name == "test_function"
    assert response.messages[0].contents[0].arguments == '{"arg1": "value1"}'
    assert response.messages[0].contents[0].call_id == "1"
    assert response.messages[1].role == "tool"
    assert response.messages[1].contents[0].type == "function_result"
    assert response.messages[1].contents[0].call_id == "1"
    assert response.messages[1].contents[0].result == "Processed value1"
    assert response.messages[2].role == "assistant"
    assert response.messages[2].text == "done"


async def test_base_client_with_function_calling_string_input(chat_client_base: SupportsChatGetResponse):
    exec_counter = 0

    @tool(name="test_function", approval_mode="never_require")
    def ai_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Processed {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="test_function", arguments='{"arg1": "value1"}')
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    response = await chat_client_base.get_response("hello", options={"tool_choice": "auto", "tools": [ai_func]})

    assert exec_counter == 1
    assert len(response.messages) == 3
    assert response.messages[1].role == "tool"
    assert response.messages[1].contents[0].type == "function_result"
    assert response.messages[1].contents[0].result == "Processed value1"


@pytest.mark.parametrize("max_iterations", [3])
async def test_base_client_with_function_calling_resets(chat_client_base: SupportsChatGetResponse):
    exec_counter = 0

    @tool(name="test_function", approval_mode="never_require")
    def ai_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Processed {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="test_function", arguments='{"arg1": "value1"}')
                ],
            )
        ),
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="2", name="test_function", arguments='{"arg1": "value1"}')
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]
    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [ai_func]}
    )
    assert exec_counter == 2
    assert len(response.messages) == 5
    assert response.messages[0].role == "assistant"
    assert response.messages[1].role == "tool"
    assert response.messages[2].role == "assistant"
    assert response.messages[3].role == "tool"
    assert response.messages[4].role == "assistant"
    assert response.messages[0].contents[0].type == "function_call"
    assert response.messages[1].contents[0].type == "function_result"
    assert response.messages[2].contents[0].type == "function_call"
    assert response.messages[3].contents[0].type == "function_result"


async def test_function_loop_applies_compaction_projection_each_model_call(chat_client_base: SupportsChatGetResponse):
    @tool(name="test_function", approval_mode="never_require")
    def ai_func(arg1: str) -> str:
        return f"Processed {arg1}"

    class _ExcludeOldestGroupAfterFirstTurn:
        async def __call__(self, messages: list[Message]) -> bool:
            groups = annotate_message_groups(messages)
            if len(groups) <= 1:
                return False
            oldest_group_id = groups[0]
            changed = False
            for message in messages:
                if _group_id(message) == oldest_group_id:
                    if message.additional_properties.get(EXCLUDED_KEY) is not True:
                        changed = True
                    message.additional_properties[EXCLUDED_KEY] = True
            return changed

    captured_roles: list[list[str]] = []
    original = chat_client_base._get_non_streaming_response  # type: ignore[attr-defined]

    async def _capture(
        *,
        messages: list[Message],
        options: dict[str, Any],
        **kwargs: Any,
    ) -> ChatResponse:
        captured_roles.append([message.role for message in messages])
        return await original(messages=messages, options=options, **kwargs)

    chat_client_base._get_non_streaming_response = _capture  # type: ignore[attr-defined,method-assign]
    chat_client_base.compaction_strategy = _ExcludeOldestGroupAfterFirstTurn()  # type: ignore[attr-defined]

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="test_function", arguments='{"arg1": "value1"}')
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [ai_func]}
    )

    assert len(captured_roles) >= 2
    assert "user" in captured_roles[0]
    assert "user" not in captured_roles[1]


async def test_function_loop_token_budget_strategy_caps_tokens_each_iteration(
    chat_client_base: SupportsChatGetResponse,
):
    exec_counter = 0
    token_budget = 500
    tokenizer = CharacterEstimatorTokenizer()

    @tool(name="test_function", approval_mode="never_require")
    def ai_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Processed {arg1}. " + ("result " * 120)

    captured_token_counts: list[int] = []
    original = chat_client_base._get_non_streaming_response  # type: ignore[attr-defined]

    async def _capture(
        *,
        messages: list[Message],
        options: dict[str, Any],
        **kwargs: Any,
    ) -> ChatResponse:
        annotate_message_groups(messages, force_reannotate=True, tokenizer=tokenizer)
        captured_token_counts.append(included_token_count(messages))
        return await original(messages=messages, options=options, **kwargs)

    chat_client_base._get_non_streaming_response = _capture  # type: ignore[attr-defined,method-assign]
    chat_client_base.tokenizer = tokenizer  # type: ignore[attr-defined]
    chat_client_base.function_invocation_configuration["max_iterations"] = 3  # type: ignore[attr-defined]
    chat_client_base.compaction_strategy = TokenBudgetComposedStrategy(  # type: ignore[attr-defined]
        token_budget=token_budget,
        tokenizer=tokenizer,
        strategies=[SlidingWindowStrategy(keep_last_groups=2)],
    )
    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="test_function", arguments='{"arg1": "value1"}')
                ],
            )
        ),
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="2", name="test_function", arguments='{"arg1": "value2"}')
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello " * 160])],
        options={"tool_choice": "auto", "tools": [ai_func]},
    )

    assert response.messages[-1].text == "done"
    assert exec_counter == 2
    assert len(captured_token_counts) >= 3
    assert all(token_count > 0 for token_count in captured_token_counts)
    assert all(token_count <= token_budget for token_count in captured_token_counts)


async def test_base_client_with_streaming_function_calling(chat_client_base: SupportsChatGetResponse):
    exec_counter = 0

    @tool(name="test_function", approval_mode="never_require")
    def ai_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Processed {arg1}"

    chat_client_base.streaming_responses = [
        [
            ChatResponseUpdate(
                contents=[Content.from_function_call(call_id="1", name="test_function", arguments='{"arg1":')],
                role="assistant",
            ),
            ChatResponseUpdate(
                contents=[Content.from_function_call(call_id="1", name="test_function", arguments='"value1"}')],
                role="assistant",
            ),
        ],
        [
            ChatResponseUpdate(
                contents=[Content.from_text(text="Processed value1")],
                role="assistant",
            )
        ],
    ]
    updates = []
    async for update in chat_client_base.get_response(
        "hello", options={"tool_choice": "auto", "tools": [ai_func]}, stream=True
    ):
        updates.append(update)
    assert len(updates) == 4  # two updates with the function call, the function result and the final text
    assert updates[0].contents[0].call_id == "1"
    assert updates[1].contents[0].call_id == "1"
    assert updates[2].contents[0].call_id == "1"
    assert updates[3].text == "Processed value1"
    assert exec_counter == 1


async def test_base_client_executes_function_calls_across_multiple_response_messages(
    chat_client_base: SupportsChatGetResponse,
):
    exec_counter = 0

    @tool(name="test_function", approval_mode="never_require")
    def ai_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Processed {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=[
                Message(
                    role="assistant",
                    contents=[
                        Content.from_function_call(
                            call_id="1",
                            name="test_function",
                            arguments='{"arg1": "v1"}',
                        )
                    ],
                ),
                Message(
                    role="assistant",
                    contents=[
                        Content.from_function_call(
                            call_id="2",
                            name="test_function",
                            arguments='{"arg1": "v2"}',
                        )
                    ],
                ),
            ],
            conversation_id="conv_after_first_call",
        ),
        ChatResponse(
            messages=Message(role="assistant", contents=["done"]),
            conversation_id="conv_after_second_call",
        ),
    ]

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])],
        options={"tool_choice": "auto", "tools": [ai_func], "conversation_id": "conv_initial"},
    )

    assert exec_counter == 2
    function_results = [
        content for msg in response.messages for content in msg.contents if content.type == "function_result"
    ]
    assert len(function_results) == 2
    assert {result.call_id for result in function_results} == {"1", "2"}


async def test_function_invocation_inside_aiohttp_server(chat_client_base: SupportsChatGetResponse):
    import aiohttp
    from aiohttp import web

    exec_counter = 0

    @tool(name="start_todo_investigation", approval_mode="never_require")
    def ai_func(user_query: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Investigated {user_query}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(
                        call_id="1",
                        name="start_todo_investigation",
                        arguments='{"user_query": "issue"}',
                    )
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    agent = Agent(client=chat_client_base, tools=[ai_func])

    async def handler(request: web.Request) -> web.Response:
        session = agent.create_session()
        result = await agent.run("Fix issue", session=session)
        return web.Response(text=result.text or "")

    app = web.Application()
    app.add_routes([web.post("/run", handler)])

    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, "127.0.0.1", 0)
    await site.start()
    try:
        port = site._server.sockets[0].getsockname()[1]
        async with aiohttp.ClientSession() as session, session.post(f"http://127.0.0.1:{port}/run") as response:
            assert response.status == 200
            await response.text()
    finally:
        await runner.cleanup()

    assert exec_counter == 1


async def test_function_invocation_in_threaded_aiohttp_app(chat_client_base: SupportsChatGetResponse):
    import asyncio
    import threading
    from queue import Queue

    import aiohttp
    from aiohttp import web

    exec_counter = 0

    @tool(name="start_threaded_investigation", approval_mode="never_require")
    def ai_func(user_query: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Threaded {user_query}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(
                        call_id="thread-1",
                        name="start_threaded_investigation",
                        arguments='{"user_query": "issue"}',
                    )
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    agent = Agent(client=chat_client_base, tools=[ai_func])

    ready_event = threading.Event()
    port_queue: Queue[int] = Queue()
    shutdown_queue: Queue[tuple[asyncio.AbstractEventLoop, asyncio.Event]] = Queue()

    async def init_app() -> web.Application:
        async def handler(request: web.Request) -> web.Response:
            session = agent.create_session()
            result = await agent.run("Fix issue", session=session)
            return web.Response(text=result.text or "")

        app = web.Application()
        app.add_routes([web.post("/run", handler)])
        return app

    def server_thread() -> None:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

        async def runner_main() -> None:
            app = await init_app()
            runner = web.AppRunner(app)
            await runner.setup()
            site = web.TCPSite(runner, "127.0.0.1", 0)
            await site.start()
            shutdown_event = asyncio.Event()
            shutdown_queue.put((loop, shutdown_event))
            port = site._server.sockets[0].getsockname()[1]
            port_queue.put(port)
            ready_event.set()
            try:
                await shutdown_event.wait()
            finally:
                await runner.cleanup()

        try:
            loop.run_until_complete(runner_main())
        finally:
            loop.close()

    thread = threading.Thread(target=server_thread, daemon=True)
    thread.start()
    ready_event.wait(timeout=5)
    assert ready_event.is_set()
    loop_ref, shutdown_event = shutdown_queue.get(timeout=2)
    port = port_queue.get(timeout=2)

    async with aiohttp.ClientSession() as session, session.post(f"http://127.0.0.1:{port}/run") as response:
        assert response.status == 200
        await response.text()

    loop_ref.call_soon_threadsafe(shutdown_event.set)
    thread.join(timeout=5)
    assert exec_counter == 1


@pytest.mark.parametrize(
    "approval_required,num_functions",
    [
        pytest.param(False, 1, id="single function without approval"),
        pytest.param(True, 1, id="single function with approval"),
        pytest.param("mixed", 2, id="two functions with mixed approval"),
    ],
)
@pytest.mark.parametrize(
    "thread_type",
    [
        pytest.param(None, id="no thread"),
        pytest.param("local", id="local thread"),
        pytest.param("service", id="service thread"),
    ],
)
@pytest.mark.parametrize("streaming", [False, True], ids=["non-streaming", "streaming"])
async def test_function_invocation_scenarios(
    chat_client_base: SupportsChatGetResponse,
    streaming: bool,
    thread_type: str | None,
    approval_required: bool | str,
    num_functions: int,
):
    """Comprehensive test for function invocation scenarios.

    This test covers:
    - Single function without approval: 3 messages (call, result, final)
    - Single function with approval: 2 messages (call, approval request)
    - Two functions with mixed approval: varies based on approval flow
    - All scenarios tested with both streaming and non-streaming
    - Thread scenarios: no thread, local thread (in-memory), and service thread (conversation_id)
    """
    exec_counter = 0

    # Setup thread based on parameters
    conversation_id = None
    if thread_type == "service":
        # Simulate a service-side thread with conversation_id
        conversation_id = "test-thread-123"

    @tool(name="no_approval_func", approval_mode="never_require")
    def func_no_approval(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Processed {arg1}"

    @tool(name="approval_func", approval_mode="always_require")
    def func_with_approval(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Approved {arg1}"

    # Setup tools and responses based on the scenario
    if num_functions == 1:
        tools = [func_with_approval if approval_required else func_no_approval]
        function_name = "approval_func" if approval_required else "no_approval_func"

        # Single function call content
        func_call = Content.from_function_call(call_id="1", name=function_name, arguments='{"arg1": "value1"}')
        completion = Message(role="assistant", contents=["done"])

        chat_client_base.run_responses = [ChatResponse(messages=Message(role="assistant", contents=[func_call]))] + (
            [] if approval_required else [ChatResponse(messages=completion)]
        )

        chat_client_base.streaming_responses = [
            [
                ChatResponseUpdate(
                    contents=[Content.from_function_call(call_id="1", name=function_name, arguments='{"arg1":')],
                    role="assistant",
                ),
                ChatResponseUpdate(
                    contents=[Content.from_function_call(call_id="1", name=function_name, arguments='"value1"}')],
                    role="assistant",
                ),
            ]
        ] + (
            []
            if approval_required
            else [[ChatResponseUpdate(contents=[Content.from_text(text="done")], role="assistant")]]
        )

    else:  # num_functions == 2
        tools = [func_no_approval, func_with_approval]

        # Two function calls content
        func_calls = [
            Content.from_function_call(call_id="1", name="no_approval_func", arguments='{"arg1": "value1"}'),
            Content.from_function_call(call_id="2", name="approval_func", arguments='{"arg1": "value2"}'),
        ]

        chat_client_base.run_responses = [ChatResponse(messages=Message(role="assistant", contents=func_calls))]

        chat_client_base.streaming_responses = [
            [
                ChatResponseUpdate(contents=[func_calls[0]], role="assistant"),
                ChatResponseUpdate(contents=[func_calls[1]], role="assistant"),
            ]
        ]

    # Execute the test
    options: dict[str, Any] = {"tool_choice": "auto", "tools": tools}
    if thread_type == "service":
        # For service threads, we need to pass conversation_id via options
        options["store"] = True
        options["conversation_id"] = conversation_id

    if not streaming:
        response = await chat_client_base.get_response([Message(role="user", contents=["hello"])], options=options)
        messages = response.messages
    else:
        updates = []
        async for update in chat_client_base.get_response(
            [Message(role="user", contents=["hello"])], options=options, stream=True
        ):
            updates.append(update)
        messages = updates

    # Service threads have different message management behavior (server-side storage)
    # so we skip detailed message assertions for those scenarios
    if thread_type == "service":
        # Just verify the function was executed or not based on approval
        if not approval_required or approval_required == "mixed":
            # For service threads, the execution counter check is still valid
            pass
        return

    # Verify based on scenario (for no thread and local thread cases)
    if num_functions == 1:
        if approval_required:
            # Single function with approval: assistant message contains both call + approval request
            if not streaming:
                assert len(messages) == 1
                # Assistant message should have FunctionCallContent + FunctionApprovalRequestContent
                assert len(messages[0].contents) == 2
                assert messages[0].contents[0].type == "function_call"
                assert messages[0].contents[1].type == "function_approval_request"
                assert messages[0].contents[1].function_call.name == "approval_func"
                assert exec_counter == 0  # Function not executed yet
            else:
                # Streaming: 2 function call chunks + 1 approval request update (same assistant message)
                assert len(messages) == 3
                assert messages[0].contents[0].type == "function_call"
                assert messages[1].contents[0].type == "function_call"
                assert messages[2].contents[0].type == "function_approval_request"
                assert messages[2].contents[0].function_call.name == "approval_func"
                assert exec_counter == 0  # Function not executed yet
        else:
            # Single function without approval: call + result + final
            if not streaming:
                assert len(messages) == 3
                assert messages[0].contents[0].type == "function_call"
                assert messages[1].contents[0].type == "function_result"
                assert messages[1].contents[0].result == "Processed value1"
                assert messages[2].role == "assistant"
                assert messages[2].text == "done"
                assert exec_counter == 1
            else:
                # Streaming has: 2 function call updates + 1 result update + 1 final update
                assert len(messages) == 4
                assert messages[0].contents[0].type == "function_call"
                assert messages[1].contents[0].type == "function_call"
                assert messages[2].contents[0].type == "function_result"
                assert messages[3].text == "done"
                assert exec_counter == 1
    else:  # num_functions == 2
        # Two functions with mixed approval
        if not streaming:
            # Mixed: assistant message has both calls + approval requests (4 items total)
            # (because when one requires approval, all are batched for approval)
            assert len(messages) == 1
            # Should have: 2 FunctionCallContent + 2 FunctionApprovalRequestContent
            assert len(messages[0].contents) == 4
            assert messages[0].contents[0].type == "function_call"
            assert messages[0].contents[1].type == "function_call"
            # Both should result in approval requests
            approval_requests = [c for c in messages[0].contents if c.type == "function_approval_request"]
            assert len(approval_requests) == 2
            assert exec_counter == 0  # Neither function executed yet
        else:
            # Streaming: 2 function call updates + 1 approval request with 2 contents
            assert len(messages) == 3
            assert messages[0].contents[0].type == "function_call"
            assert messages[1].contents[0].type == "function_call"
            # The approval request message contains both approval requests
            assert len(messages[2].contents) == 2
            assert all(c.type == "function_approval_request" for c in messages[2].contents)
            assert exec_counter == 0  # Neither function executed yet


async def test_rejected_approval(chat_client_base: SupportsChatGetResponse):
    """Test that rejecting an approval alongside an approved one is handled correctly."""

    exec_counter_approved = 0
    exec_counter_rejected = 0

    @tool(name="approved_func", approval_mode="always_require")
    def func_approved(arg1: str) -> str:
        nonlocal exec_counter_approved
        exec_counter_approved += 1
        return f"Approved {arg1}"

    @tool(name="rejected_func", approval_mode="always_require")
    def func_rejected(arg1: str) -> str:
        nonlocal exec_counter_rejected
        exec_counter_rejected += 1
        return f"Rejected {arg1}"

    # Setup: two function calls that require approval
    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="approved_func", arguments='{"arg1": "value1"}'),
                    Content.from_function_call(call_id="2", name="rejected_func", arguments='{"arg1": "value2"}'),
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    # Get the response with approval requests
    response = await chat_client_base.get_response(
        "hello", options={"tool_choice": "auto", "tools": [func_approved, func_rejected]}
    )
    # Approval requests are now added to the assistant message, not a separate message
    assert len(response.messages) == 1
    # Assistant message should have: 2 FunctionCallContent + 2 FunctionApprovalRequestContent
    assert len(response.messages[0].contents) == 4
    approval_requests = [c for c in response.messages[0].contents if c.type == "function_approval_request"]
    assert len(approval_requests) == 2

    # Approve one and reject the other
    approval_req_1 = approval_requests[0]
    approval_req_2 = approval_requests[1]

    approved_response = Content.from_function_approval_response(
        id=approval_req_1.id,
        function_call=approval_req_1.function_call,
        approved=True,
    )
    rejected_response = Content.from_function_approval_response(
        id=approval_req_2.id,
        function_call=approval_req_2.function_call,
        approved=False,
    )

    # Continue conversation with one approved and one rejected
    all_messages = response.messages + [Message(role="user", contents=[approved_response, rejected_response])]

    # Call get_response which will process the approvals
    await chat_client_base.get_response(
        all_messages, options={"tool_choice": "auto", "tools": [func_approved, func_rejected]}
    )

    # Verify the approval/rejection was processed correctly
    # Find the results in the input messages (modified in-place)
    approved_result = None
    rejected_result = None
    for msg in all_messages:
        for content in msg.contents:
            if content.type == "function_result":
                if content.call_id == "1":
                    approved_result = content
                elif content.call_id == "2":
                    rejected_result = content

    # The approved function should have been executed and have a result
    assert approved_result is not None, "Should have found result for approved function"
    assert approved_result.result == "Approved value1"
    assert exec_counter_approved == 1

    # The rejected function should have a "not approved" result and NOT have been executed
    assert rejected_result is not None, "Should have found result for rejected function"
    assert rejected_result.result == "Error: Tool call invocation was rejected by user."
    assert exec_counter_rejected == 0

    # Verify that messages with FunctionResultContent have role="tool"
    # This ensures the message format is correct for OpenAI's API
    for msg in all_messages:
        for content in msg.contents:
            if content.type == "function_result":
                assert msg.role == "tool", f"Message with FunctionResultContent must have role='tool', got '{msg.role}'"


async def test_approval_requests_in_assistant_message(chat_client_base: SupportsChatGetResponse):
    """Approval requests should be added to the assistant message that contains the function call."""
    exec_counter = 0

    @tool(name="test_func", approval_mode="always_require")
    def func_with_approval(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Result {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="test_func", arguments='{"arg1": "value1"}'),
                ],
            )
        ),
    ]

    response = await chat_client_base.get_response(
        "hello", options={"tool_choice": "auto", "tools": [func_with_approval]}
    )

    # Should have one assistant message containing both the call and approval request
    assert len(response.messages) == 1
    assert response.messages[0].role == "assistant"
    assert len(response.messages[0].contents) == 2
    assert response.messages[0].contents[0].type == "function_call"
    assert response.messages[0].contents[1].type == "function_approval_request"
    assert exec_counter == 0


async def test_persisted_approval_messages_replay_correctly(chat_client_base: SupportsChatGetResponse):
    """Approval flow should work when messages are persisted and sent back (thread scenario)."""

    exec_counter = 0

    @tool(name="test_func", approval_mode="always_require")
    def func_with_approval(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Result {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="test_func", arguments='{"arg1": "value1"}'),
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    # Get approval request
    response1 = await chat_client_base.get_response(
        "hello", options={"tool_choice": "auto", "tools": [func_with_approval]}
    )

    # Store messages (like a thread would)
    persisted_messages = [
        Message(role="user", contents=["hello"]),
        *response1.messages,
    ]

    # Send approval
    approval_req = [c for c in response1.messages[0].contents if c.type == "function_approval_request"][0]
    approval_response = Content.from_function_approval_response(
        id=approval_req.id,
        function_call=approval_req.function_call,
        approved=True,
    )
    persisted_messages.append(Message(role="user", contents=[approval_response]))

    # Continue with all persisted messages
    response2 = await chat_client_base.get_response(
        persisted_messages, options={"tool_choice": "auto", "tools": [func_with_approval]}
    )

    # Should execute successfully
    assert response2 is not None
    assert exec_counter == 1


async def test_no_duplicate_function_calls_after_approval_processing(chat_client_base: SupportsChatGetResponse):
    """Processing approval should not create duplicate function calls in messages."""

    @tool(name="test_func", approval_mode="always_require")
    def func_with_approval(arg1: str) -> str:
        return f"Result {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="test_func", arguments='{"arg1": "value1"}'),
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    response1 = await chat_client_base.get_response(
        "hello", options={"tool_choice": "auto", "tools": [func_with_approval]}
    )

    approval_req = [c for c in response1.messages[0].contents if c.type == "function_approval_request"][0]
    approval_response = Content.from_function_approval_response(
        id=approval_req.id,
        function_call=approval_req.function_call,
        approved=True,
    )

    all_messages = response1.messages + [Message(role="user", contents=[approval_response])]
    await chat_client_base.get_response(all_messages, options={"tool_choice": "auto", "tools": [func_with_approval]})

    # Count function calls with the same call_id
    function_call_count = sum(
        1
        for msg in all_messages
        for content in msg.contents
        if content.type == "function_call" and content.call_id == "1"
    )

    assert function_call_count == 1


async def test_rejection_result_uses_function_call_id(chat_client_base: SupportsChatGetResponse):
    """Rejection error result should use the function call's call_id, not the approval's id."""

    @tool(name="test_func", approval_mode="always_require")
    def func_with_approval(arg1: str) -> str:
        return f"Result {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="call_123", name="test_func", arguments='{"arg1": "value1"}'),
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    response1 = await chat_client_base.get_response(
        "hello", options={"tool_choice": "auto", "tools": [func_with_approval]}
    )

    approval_req = [c for c in response1.messages[0].contents if c.type == "function_approval_request"][0]
    rejection_response = Content.from_function_approval_response(
        id=approval_req.id,
        function_call=approval_req.function_call,
        approved=False,
    )

    all_messages = response1.messages + [Message(role="user", contents=[rejection_response])]
    await chat_client_base.get_response(all_messages, options={"tool_choice": "auto", "tools": [func_with_approval]})

    # Find the rejection result
    rejection_result = next(
        (content for msg in all_messages for content in msg.contents if content.type == "function_result"),
        None,
    )

    assert rejection_result is not None
    assert rejection_result.call_id == "call_123"
    assert "rejected" in rejection_result.result.lower()


async def test_max_iterations_limit(chat_client_base: SupportsChatGetResponse):
    """Test that MAX_ITERATIONS in additional_properties limits function call loops."""
    exec_counter = 0

    @tool(name="test_function", approval_mode="never_require")
    def ai_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Processed {arg1}"

    # Set up multiple function call responses to create a loop
    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="test_function", arguments='{"arg1": "value1"}')
                ],
            )
        ),
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="2", name="test_function", arguments='{"arg1": "value2"}')
                ],
            )
        ),
        # Failsafe response when tool_choice is set to "none"
        ChatResponse(messages=Message(role="assistant", contents=["giving up on tools"])),
    ]

    # Set max_iterations to 1 in additional_properties
    chat_client_base.function_invocation_configuration["max_iterations"] = 1

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [ai_func]}
    )

    # With max_iterations=1, we should:
    # 1. Execute first function call (exec_counter=1)
    # 2. Try to make second call but hit iteration limit
    # 3. Fall back to asking for a plain answer with tool_choice="none"
    assert exec_counter == 1  # Only first function executed
    assert response.messages[-1].text == "I broke out of the function invocation loop..."  # Failsafe response


async def test_max_iterations_no_orphaned_function_calls(chat_client_base: SupportsChatGetResponse):
    """When max_iterations is reached, verify the returned response has no orphaned
    FunctionCallContent (i.e., every function_call has a matching function_result).
    """
    exec_counter = 0

    @tool(name="test_function", approval_mode="never_require")
    def ai_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Processed {arg1}"

    # Model keeps requesting tool calls on every iteration
    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="call_1", name="test_function", arguments='{"arg1": "v1"}')
                ],
            )
        ),
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="call_2", name="test_function", arguments='{"arg1": "v2"}')
                ],
            )
        ),
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="call_3", name="test_function", arguments='{"arg1": "v3"}')
                ],
            )
        ),
    ]

    chat_client_base.function_invocation_configuration["max_iterations"] = 2

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])],
        options={"tool_choice": "auto", "tools": [ai_func]},
    )

    # Collect all function_call and function_result call_ids from response
    all_call_ids = set()
    all_result_ids = set()
    for msg in response.messages:
        for content in msg.contents:
            if content.type == "function_call":
                all_call_ids.add(content.call_id)
            elif content.type == "function_result":
                all_result_ids.add(content.call_id)

    orphaned_calls = all_call_ids - all_result_ids
    assert not orphaned_calls, (
        f"Response contains orphaned FunctionCallContent without matching FunctionResultContent: {orphaned_calls}."
    )


async def test_max_iterations_makes_final_toolchoice_none_call(chat_client_base: SupportsChatGetResponse):
    """When max_iterations is reached, verify a final model call is made with
    tool_choice='none' to produce a clean text response.
    """
    exec_counter = 0

    @tool(name="test_function", approval_mode="never_require")
    def ai_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Processed {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="call_1", name="test_function", arguments='{"arg1": "v1"}')
                ],
            )
        ),
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="call_2", name="test_function", arguments='{"arg1": "v2"}')
                ],
            )
        ),
        # This response should be reached via failsafe (tool_choice="none")
        ChatResponse(messages=Message(role="assistant", contents=["Final answer after giving up on tools."])),
    ]

    chat_client_base.function_invocation_configuration["max_iterations"] = 1

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])],
        options={"tool_choice": "auto", "tools": [ai_func]},
    )

    assert exec_counter == 1, f"Expected 1 function execution, got {exec_counter}"

    # The response should end with a plain text message (from the failsafe call)
    last_msg = response.messages[-1]
    has_function_calls = any(c.type == "function_call" for c in last_msg.contents)

    assert not has_function_calls, (
        f"Last message in response still contains function_call items. "
        f"Expected a clean text response after max_iterations failsafe. "
        f"Got message with role={last_msg.role}, contents={[c.type for c in last_msg.contents]}"
    )

    # The mock client returns "I broke out of the function invocation loop..."
    # when tool_choice="none"
    assert last_msg.text == "I broke out of the function invocation loop...", (
        f"Expected failsafe text response, got: {last_msg.text!r}"
    )


async def test_max_iterations_preserves_all_fcc_messages(chat_client_base: SupportsChatGetResponse):
    """When max_iterations is reached and a final response is produced, all
    intermediate function call/result messages should be included.
    """
    exec_counter = 0

    @tool(name="test_function", approval_mode="never_require")
    def ai_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Result {exec_counter}"

    # Two iterations of function calls, then failsafe
    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="call_1", name="test_function", arguments='{"arg1": "v1"}')
                ],
            )
        ),
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="call_2", name="test_function", arguments='{"arg1": "v2"}')
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["Done"])),
    ]

    chat_client_base.function_invocation_configuration["max_iterations"] = 2

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])],
        options={"tool_choice": "auto", "tools": [ai_func]},
    )

    assert exec_counter == 2, f"Expected 2 function executions, got {exec_counter}"

    # All function calls from both iterations should be present in the response
    all_call_ids = set()
    all_result_ids = set()
    for msg in response.messages:
        for content in msg.contents:
            if content.type == "function_call":
                all_call_ids.add(content.call_id)
            elif content.type == "function_result":
                all_result_ids.add(content.call_id)

    assert "call_1" in all_call_ids, "First iteration's function call missing from response"
    assert "call_2" in all_call_ids, "Second iteration's function call missing from response"

    assert all_call_ids == all_result_ids, (
        f"Mismatched function calls and results. Calls: {all_call_ids}, Results: {all_result_ids}"
    )


async def test_max_iterations_thread_integrity_with_agent(chat_client_base: SupportsChatGetResponse):
    """Verify that agent.run() does not produce orphaned function calls after
    max_iterations, which would corrupt the thread and cause API errors on the
    next call.
    """

    @tool(name="browser_snapshot", approval_mode="never_require")
    def browser_snapshot(url: str) -> str:
        return f"Screenshot of {url}"

    # Model keeps requesting tool calls on every iteration.
    # The failsafe call (with tool_choice="none") after the loop is handled
    # automatically by the mock client, which returns a hardcoded text response
    # when tool_choice="none" (see conftest.py ChatClientBase.get_response).
    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(
                        call_id="call_abc", name="browser_snapshot", arguments='{"url": "https://example.com"}'
                    )
                ],
            )
        ),
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(
                        call_id="call_xyz", name="browser_snapshot", arguments='{"url": "https://test.com"}'
                    )
                ],
            )
        ),
    ]

    chat_client_base.function_invocation_configuration["max_iterations"] = 2

    agent = Agent(
        client=chat_client_base,
        name="test-agent",
        tools=[browser_snapshot],
    )

    response = await agent.run(
        "Take screenshots",
        options={"tool_choice": "auto"},
    )

    # Check for orphaned function calls in the response messages
    all_call_ids = set()
    all_result_ids = set()
    for msg in response.messages:
        for content in msg.contents:
            if content.type == "function_call":
                all_call_ids.add(content.call_id)
            elif content.type == "function_result":
                all_result_ids.add(content.call_id)

    orphaned_calls = all_call_ids - all_result_ids
    assert not orphaned_calls, (
        f"Response contains orphaned function calls {orphaned_calls}. This would cause API errors on the next call."
    )


@pytest.mark.parametrize("max_iterations", [10])
async def test_max_function_calls_limits_parallel_invocations(chat_client_base: SupportsChatGetResponse):
    """Test that max_function_calls caps total function invocations across iterations with parallel calls."""
    exec_counter = 0

    @tool(name="search", approval_mode="never_require")
    def search_func(query: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Result for {query}"

    # Each iteration returns 3 parallel tool calls
    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1a", name="search", arguments='{"query": "q1"}'),
                    Content.from_function_call(call_id="1b", name="search", arguments='{"query": "q2"}'),
                    Content.from_function_call(call_id="1c", name="search", arguments='{"query": "q3"}'),
                ],
            )
        ),
        # Second iteration: 3 more parallel calls (total would be 6, exceeding limit of 5)
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="2a", name="search", arguments='{"query": "q4"}'),
                    Content.from_function_call(call_id="2b", name="search", arguments='{"query": "q5"}'),
                    Content.from_function_call(call_id="2c", name="search", arguments='{"query": "q6"}'),
                ],
            )
        ),
        # Final response after tool_choice="none" is forced
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    # Allow many iterations but cap total function calls at 5
    chat_client_base.function_invocation_configuration["max_function_calls"] = 5

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["search"])], options={"tool_choice": "auto", "tools": [search_func]}
    )

    # First iteration executes 3 calls (total=3, under limit).
    # Second iteration executes 3 more (total=6, reaches limit) then forces tool_choice="none".
    # The loop completes the current batch before stopping.
    assert exec_counter == 6
    assert "broke out" in response.messages[-1].text


@pytest.mark.parametrize("max_iterations", [10])
async def test_max_function_calls_single_calls_per_iteration(chat_client_base: SupportsChatGetResponse):
    """Test that max_function_calls works with single tool calls per iteration."""
    exec_counter = 0

    @tool(name="lookup", approval_mode="never_require")
    def lookup_func(key: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Value for {key}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="lookup", arguments='{"key": "a"}'),
                ],
            )
        ),
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="2", name="lookup", arguments='{"key": "b"}'),
                ],
            )
        ),
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="3", name="lookup", arguments='{"key": "c"}'),
                ],
            )
        ),
        # After limit is reached
        ChatResponse(messages=Message(role="assistant", contents=["all done"])),
    ]

    chat_client_base.function_invocation_configuration["max_function_calls"] = 2

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["look up keys"])], options={"tool_choice": "auto", "tools": [lookup_func]}
    )

    # 2 single calls executed, then limit reached, tool_choice="none" forced
    assert exec_counter == 2
    assert "broke out" in response.messages[-1].text


@pytest.mark.parametrize("max_iterations", [10])
async def test_max_function_calls_none_means_unlimited(chat_client_base: SupportsChatGetResponse):
    """Test that max_function_calls=None (default) allows unlimited function calls."""
    exec_counter = 0

    @tool(name="do_thing", approval_mode="never_require")
    def do_thing_func(arg: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Done {arg}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id=str(i), name="do_thing", arguments=f'{{"arg": "v{i}"}}'),
                ],
            )
        )
        for i in range(5)
    ] + [ChatResponse(messages=Message(role="assistant", contents=["finished"]))]

    # Explicitly set to None (default) — should not limit
    chat_client_base.function_invocation_configuration["max_function_calls"] = None

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["do things"])], options={"tool_choice": "auto", "tools": [do_thing_func]}
    )

    assert exec_counter == 5
    assert response.messages[-1].text == "finished"


async def test_function_invocation_config_enabled_false(chat_client_base: SupportsChatGetResponse):
    """Test that setting enabled=False disables function invocation."""
    exec_counter = 0

    @tool(name="test_function")
    def ai_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Processed {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(messages=Message(role="assistant", contents=["response without function calling"])),
    ]

    # Disable function invocation
    chat_client_base.function_invocation_configuration["enabled"] = False

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [ai_func]}
    )

    # Function should not be executed - when enabled=False, the loop doesn't run
    assert exec_counter == 0
    # The response should be from the mock client
    assert len(response.messages) > 0


async def test_function_invocation_config_enabled_false_preserves_invocation_kwargs(
    chat_client_base: SupportsChatGetResponse,
):
    """Test disabled function invocation still forwards invocation kwargs downstream."""
    captured_kwargs: dict[str, Any] = {}

    @tool(name="test_function")
    def ai_func(arg1: str) -> str:
        return f"Processed {arg1}"

    @chat_middleware
    async def capture_middleware(context, call_next):
        captured_kwargs.update(context.function_invocation_kwargs or {})
        await call_next()

    chat_client_base.chat_middleware = [capture_middleware]
    chat_client_base.run_responses = [
        ChatResponse(messages=Message(role="assistant", contents=["response without function calling"])),
    ]
    chat_client_base.function_invocation_configuration["enabled"] = False

    await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])],
        options={"tool_choice": "auto", "tools": [ai_func]},
        function_invocation_kwargs={"tool_request_id": "tool-123"},
    )

    assert captured_kwargs == {"tool_request_id": "tool-123"}


@pytest.mark.skip(reason="Error handling and failsafe behavior needs investigation in unified API")
async def test_function_invocation_config_max_consecutive_errors(chat_client_base: SupportsChatGetResponse):
    """Test that max_consecutive_errors_per_request limits error retries."""

    @tool(name="error_function", approval_mode="never_require")
    def error_func(arg1: str) -> str:
        raise ValueError("Function error")

    # Set up multiple function call responses that will all error
    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="error_function", arguments='{"arg1": "value1"}')
                ],
            )
        ),
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="2", name="error_function", arguments='{"arg1": "value2"}')
                ],
            )
        ),
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="3", name="error_function", arguments='{"arg1": "value3"}')
                ],
            )
        ),
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="4", name="error_function", arguments='{"arg1": "value4"}')
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["final response"])),
    ]

    # Set max_consecutive_errors to 2
    chat_client_base.function_invocation_configuration["max_consecutive_errors_per_request"] = 2

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [error_func]}
    )

    # Should stop after 2 consecutive errors and force a non-tool response
    error_results = [
        content
        for msg in response.messages
        for content in msg.contents
        if content.type == "function_result" and content.exception is not None
    ]
    # The first call errors, then the second call errors, hitting the limit
    # So we get 2 function calls with errors, but the responses show the behavior stopped
    assert len(error_results) >= 1  # At least one error occurred
    # Should have stopped making new function calls after hitting the error limit
    function_calls = [
        content for msg in response.messages for content in msg.contents if content.type == "function_call"
    ]
    # Should have made at most 2 function calls before stopping
    assert len(function_calls) <= 2


async def test_function_invocation_stop_clears_conversation_id_non_stream(chat_client_base: SupportsChatGetResponse):
    """Stop-path responses should not carry a continuation conversation_id."""

    @tool(name="error_function", approval_mode="never_require")
    def error_func(arg1: str) -> str:
        raise ValueError("Function error")

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="error_function", arguments='{"arg1": "value1"}')
                ],
            ),
            conversation_id="resp_1",
        )
    ]
    chat_client_base.function_invocation_configuration["max_consecutive_errors_per_request"] = 1
    session_stub = type("SessionStub", (), {"service_session_id": "resp_seed"})()

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])],
        options={"tool_choice": "auto", "tools": [error_func]},
        client_kwargs={"session": session_stub},
    )

    assert response.conversation_id is None


async def test_function_invocation_config_terminate_on_unknown_calls_false(chat_client_base: SupportsChatGetResponse):
    """Test that terminate_on_unknown_calls=False returns error message for unknown functions."""
    exec_counter = 0

    @tool(name="known_function")
    def known_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Processed {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="unknown_function", arguments='{"arg1": "value1"}')
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    # Set terminate_on_unknown_calls to False (default)
    chat_client_base.function_invocation_configuration["terminate_on_unknown_calls"] = False

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [known_func]}
    )

    # Should have a result message indicating the tool wasn't found
    assert len(response.messages) == 3
    assert response.messages[1].contents[0].type == "function_result"
    result_str = response.messages[1].contents[0].result or response.messages[1].contents[0].exception or ""
    assert "not found" in result_str.lower()
    assert exec_counter == 0  # Known function not executed


async def test_function_invocation_config_terminate_on_unknown_calls_true(chat_client_base: SupportsChatGetResponse):
    """Test that terminate_on_unknown_calls=True stops execution on unknown functions."""
    exec_counter = 0

    @tool(name="known_function")
    def known_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Processed {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="unknown_function", arguments='{"arg1": "value1"}')
                ],
            )
        ),
    ]

    # Set terminate_on_unknown_calls to True
    chat_client_base.function_invocation_configuration["terminate_on_unknown_calls"] = True

    # Should raise an exception when encountering an unknown function
    with pytest.raises(KeyError, match='Error: Requested function "unknown_function" not found'):
        await chat_client_base.get_response(
            [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [known_func]}
        )

    assert exec_counter == 0


async def test_function_invocation_config_additional_tools(chat_client_base: SupportsChatGetResponse):
    """Test that additional_tools are available but treated as declaration_only."""
    exec_counter_visible = 0
    exec_counter_hidden = 0

    @tool(name="visible_function")
    def visible_func(arg1: str) -> str:
        nonlocal exec_counter_visible
        exec_counter_visible += 1
        return f"Visible {arg1}"

    @tool(name="hidden_function")
    def hidden_func(arg1: str) -> str:
        nonlocal exec_counter_hidden
        exec_counter_hidden += 1
        return f"Hidden {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="hidden_function", arguments='{"arg1": "value1"}')
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    # Add hidden_func to additional_tools
    chat_client_base.function_invocation_configuration["additional_tools"] = [hidden_func]

    # Only pass visible_func in the tools parameter
    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [visible_func]}
    )

    # Additional tools are treated as declaration_only, so not executed
    # The function call should be in the messages but not executed
    assert exec_counter_hidden == 0
    assert exec_counter_visible == 0
    # Should have the function call in messages (declaration_only behavior)
    function_calls = [
        content
        for msg in response.messages
        for content in msg.contents
        if content.type == "function_call" and content.name == "hidden_function"
    ]
    assert len(function_calls) >= 1


async def test_function_invocation_config_include_detailed_errors_false(chat_client_base: SupportsChatGetResponse):
    """Test that include_detailed_errors=False returns generic error messages."""

    @tool(name="error_function", approval_mode="never_require")
    def error_func(arg1: str) -> str:
        raise ValueError("Specific error message that should not appear")

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="error_function", arguments='{"arg1": "value1"}')
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    # Set include_detailed_errors to False (default)
    chat_client_base.function_invocation_configuration["include_detailed_errors"] = False

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [error_func]}
    )

    # Should have a generic error message
    error_result = next(
        content for msg in response.messages for content in msg.contents if content.type == "function_result"
    )
    assert error_result.result is not None
    assert error_result.exception is not None
    assert "Specific error message" not in error_result.result
    assert "Error:" in error_result.result  # Generic error prefix


async def test_function_invocation_config_include_detailed_errors_true(chat_client_base: SupportsChatGetResponse):
    """Test that include_detailed_errors=True returns detailed error information."""

    @tool(name="error_function", approval_mode="never_require")
    def error_func(arg1: str) -> str:
        raise ValueError("Specific error message that should appear")

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="error_function", arguments='{"arg1": "value1"}')
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    # Set include_detailed_errors to True
    chat_client_base.function_invocation_configuration["include_detailed_errors"] = True

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [error_func]}
    )

    # Should have detailed error message
    error_result = next(
        content for msg in response.messages for content in msg.contents if content.type == "function_result"
    )
    assert error_result.result is not None
    assert error_result.exception is not None
    assert "Specific error message that should appear" in error_result.result
    # The error format includes "Function failed. Exception:" prefix
    assert "Exception:" in error_result.result


async def test_function_invocation_config_validation_max_iterations():
    """Test that max_iterations validation works correctly."""
    from agent_framework import normalize_function_invocation_configuration

    # Valid values
    config = normalize_function_invocation_configuration({"max_iterations": 1})
    assert config["max_iterations"] == 1

    config = normalize_function_invocation_configuration({"max_iterations": 100})
    assert config["max_iterations"] == 100

    # Invalid value (less than 1)
    with pytest.raises(ValueError, match="max_iterations must be at least 1"):
        normalize_function_invocation_configuration({"max_iterations": 0})

    with pytest.raises(ValueError, match="max_iterations must be at least 1"):
        normalize_function_invocation_configuration({"max_iterations": -1})


async def test_function_invocation_config_validation_max_consecutive_errors():
    """Test that max_consecutive_errors_per_request validation works correctly."""
    from agent_framework import normalize_function_invocation_configuration

    # Valid values
    config = normalize_function_invocation_configuration({"max_consecutive_errors_per_request": 0})
    assert config["max_consecutive_errors_per_request"] == 0

    config = normalize_function_invocation_configuration({"max_consecutive_errors_per_request": 5})
    assert config["max_consecutive_errors_per_request"] == 5

    # Invalid value (less than 0)
    with pytest.raises(ValueError, match="max_consecutive_errors_per_request must be 0 or more"):
        normalize_function_invocation_configuration({"max_consecutive_errors_per_request": -1})


async def test_function_invocation_config_validation_max_function_calls():
    """Test that max_function_calls validation works correctly."""
    from agent_framework import normalize_function_invocation_configuration

    # Default is None (unlimited)
    config = normalize_function_invocation_configuration(None)
    assert config["max_function_calls"] is None

    # Valid values
    config = normalize_function_invocation_configuration({"max_function_calls": 1})
    assert config["max_function_calls"] == 1

    config = normalize_function_invocation_configuration({"max_function_calls": 100})
    assert config["max_function_calls"] == 100

    # None is valid (unlimited)
    config = normalize_function_invocation_configuration({"max_function_calls": None})
    assert config["max_function_calls"] is None

    # Invalid value (less than 1)
    with pytest.raises(ValueError, match="max_function_calls must be at least 1 or None"):
        normalize_function_invocation_configuration({"max_function_calls": 0})

    with pytest.raises(ValueError, match="max_function_calls must be at least 1 or None"):
        normalize_function_invocation_configuration({"max_function_calls": -1})


async def test_argument_validation_error_with_detailed_errors(chat_client_base: SupportsChatGetResponse):
    """Test that argument validation errors include details when include_detailed_errors=True."""

    @tool(name="typed_function", approval_mode="never_require")
    def typed_func(arg1: int) -> str:  # Expects int, not str
        return f"Got {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="typed_function", arguments='{"arg1": "not_an_int"}')
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    # Set include_detailed_errors to True
    chat_client_base.function_invocation_configuration["include_detailed_errors"] = True

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [typed_func]}
    )

    # Should have detailed validation error
    error_result = next(
        content for msg in response.messages for content in msg.contents if content.type == "function_result"
    )
    assert error_result.result is not None
    assert error_result.exception is not None
    assert "Argument parsing failed" in error_result.result
    assert "Exception:" in error_result.result  # Detailed error included


async def test_argument_validation_error_without_detailed_errors(chat_client_base: SupportsChatGetResponse):
    """Test that argument validation errors are generic when include_detailed_errors=False."""

    @tool(name="typed_function", approval_mode="never_require")
    def typed_func(arg1: int) -> str:  # Expects int, not str
        return f"Got {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="typed_function", arguments='{"arg1": "not_an_int"}')
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    # Set include_detailed_errors to False (default)
    chat_client_base.function_invocation_configuration["include_detailed_errors"] = False

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [typed_func]}
    )

    # Should have generic validation error
    error_result = next(
        content for msg in response.messages for content in msg.contents if content.type == "function_result"
    )
    assert error_result.result is not None
    assert error_result.exception is not None
    assert "Argument parsing failed" in error_result.result
    assert "Exception:" not in error_result.result  # No detailed error


async def test_hosted_tool_approval_response(chat_client_base: SupportsChatGetResponse):
    """Test handling of approval responses for hosted tools (tools not in tool_map)."""

    @tool(name="local_function")
    def local_func(arg1: str) -> str:
        return f"Local {arg1}"

    # Create an approval response for a hosted tool that's not in our tool_map
    hosted_function_call = Content.from_function_call(
        call_id="hosted_1", name="hosted_function", arguments='{"arg1": "value"}'
    )
    approval_response = Content.from_function_approval_response(
        id="approval_1",
        function_call=hosted_function_call,
        approved=True,
    )

    chat_client_base.run_responses = [
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    # Send the approval response
    response = await chat_client_base.get_response(
        [Message(role="user", contents=[approval_response])],
        options={"tool_choice": "auto", "tools": [local_func]},
    )

    # The hosted tool approval should be returned as-is (not executed)
    # Check that we got a response without errors
    assert response is not None


async def test_hosted_mcp_approval_response_passthrough(chat_client_base: SupportsChatGetResponse):
    """Test that hosted MCP approval responses pass through without local execution.

    When an MCP approval response has server_label in function_call.additional_properties,
    the function invocation layer must not intercept it. The approval request/response
    should be forwarded to the API as-is so the service can execute the hosted tool.
    """

    @tool(name="local_function")
    def local_func(arg1: str) -> str:
        return f"Local {arg1}"

    # Simulate an MCP approval request from the service (has server_label)
    mcp_function_call = Content.from_function_call(
        call_id="mcpr_abc123",
        name="microsoft_docs_search",
        arguments='{"query": "azure storage"}',
        additional_properties={"server_label": "Microsoft_Learn_MCP"},
    )
    mcp_approval_request = Content.from_function_approval_request(
        id="mcpr_abc123",
        function_call=mcp_function_call,
    )
    mcp_approval_response = mcp_approval_request.to_function_approval_response(approved=True)

    # The second call (after approval) should return a final response
    chat_client_base.run_responses = [
        ChatResponse(messages=Message(role="assistant", contents=["Here are the docs results."])),
    ]

    # Build message list mimicking handle_approvals_without_session:
    # [original query, assistant with approval_request, user with approval_response]
    messages = [
        Message(role="user", contents=["Search docs for azure storage"]),
        Message(role="assistant", contents=[mcp_approval_request]),
        Message(role="user", contents=[mcp_approval_response]),
    ]

    response = await chat_client_base.get_response(
        messages,
        options={"tool_choice": "auto", "tools": [local_func]},
    )

    # The response should succeed without errors
    assert response is not None
    assert response.messages[0].text == "Here are the docs results."

    # The approval contents should NOT have been mutated by the function invocation layer.
    # The assistant message should still have the original approval_request content.
    assistant_msg = messages[1]
    assert assistant_msg.contents[0].type == "function_approval_request"
    # The user message should still have the original approval_response content.
    user_msg = messages[2]
    assert user_msg.contents[0].type == "function_approval_response"


def test_is_hosted_tool_approval_with_server_label():
    """Test that _is_hosted_tool_approval returns True for MCP approvals with server_label."""
    from agent_framework._tools import _is_hosted_tool_approval

    mcp_fc = Content.from_function_call(
        call_id="mcpr_abc",
        name="docs_search",
        arguments="{}",
        additional_properties={"server_label": "Microsoft_Learn_MCP"},
    )
    mcp_request = Content.from_function_approval_request(id="mcpr_abc", function_call=mcp_fc)
    mcp_response = mcp_request.to_function_approval_response(approved=True)

    assert _is_hosted_tool_approval(mcp_request) is True
    assert _is_hosted_tool_approval(mcp_response) is True


def test_is_hosted_tool_approval_without_server_label():
    """Test that _is_hosted_tool_approval returns False for regular tool approvals."""
    from agent_framework._tools import _is_hosted_tool_approval

    regular_fc = Content.from_function_call(call_id="call_1", name="my_func", arguments="{}")
    regular_request = Content.from_function_approval_request(id="call_1", function_call=regular_fc)
    regular_response = regular_request.to_function_approval_response(approved=True)

    assert _is_hosted_tool_approval(regular_request) is False
    assert _is_hosted_tool_approval(regular_response) is False
    # Also test with None/non-content objects
    assert _is_hosted_tool_approval(None) is False
    assert _is_hosted_tool_approval("not a content") is False


def test_replace_approval_contents_with_results_uses_result_call_ids_without_placeholders() -> None:
    from agent_framework._tools import _collect_approval_responses, _replace_approval_contents_with_results

    call_one, request_one, response_one = _build_approved_tool_roundtrip(
        call_id="call_1", approval_id="approval_1", tool_name="first_tool"
    )
    call_two, request_two, response_two = _build_approved_tool_roundtrip(
        call_id="call_2", approval_id="approval_2", tool_name="second_tool"
    )

    messages = [
        Message(role="assistant", contents=[call_one, request_one, call_two, request_two]),
        Message(role="user", contents=[response_one, response_two]),
    ]

    _replace_approval_contents_with_results(
        messages,
        _collect_approval_responses(messages),
        [
            Content.from_function_result(call_id="call_2", result="second result"),
            Content.from_function_result(call_id="call_1", result="first result"),
        ],
    )

    assert len(messages) == 2
    assert messages[0].contents == [call_one, call_two]
    assert messages[1].role == "tool"
    assert [(content.call_id, content.result) for content in messages[1].contents] == [
        ("call_1", "first result"),
        ("call_2", "second result"),
    ]


def test_replace_approval_contents_with_results_uses_result_call_ids_for_placeholders() -> None:
    from agent_framework._tools import _collect_approval_responses, _replace_approval_contents_with_results

    call_one, request_one, response_one = _build_approved_tool_roundtrip(
        call_id="call_1", approval_id="approval_1", tool_name="first_tool"
    )
    call_two, request_two, response_two = _build_approved_tool_roundtrip(
        call_id="call_2", approval_id="approval_2", tool_name="second_tool"
    )

    messages = [
        Message(role="assistant", contents=[call_one, request_one, call_two, request_two]),
        Message(
            role="tool",
            contents=[
                Content.from_function_result(call_id="call_1", result="[APPROVAL_PENDING] first placeholder"),
                Content.from_function_result(call_id="call_2", result="[APPROVAL_PENDING] second placeholder"),
            ],
        ),
        Message(role="user", contents=[response_one, response_two]),
    ]

    _replace_approval_contents_with_results(
        messages,
        _collect_approval_responses(messages),
        [
            Content.from_function_result(call_id="call_2", result="second result"),
            Content.from_function_result(call_id="call_1", result="first result"),
        ],
    )

    assert len(messages) == 2
    assert messages[0].contents == [call_one, call_two]
    assert [(content.call_id, content.result) for content in messages[1].contents] == [
        ("call_1", "first result"),
        ("call_2", "second result"),
    ]


def test_replace_approval_contents_with_results_skips_results_without_call_id() -> None:
    from agent_framework._tools import _collect_approval_responses, _replace_approval_contents_with_results

    call_one, request_one, response_one = _build_approved_tool_roundtrip(
        call_id="call_1", approval_id="approval_1", tool_name="first_tool"
    )

    messages = [
        Message(role="assistant", contents=[call_one, request_one]),
        Message(
            role="tool",
            contents=[Content.from_function_result(call_id="call_1", result="[APPROVAL_PENDING] placeholder")],
        ),
        Message(role="user", contents=[response_one]),
    ]

    _replace_approval_contents_with_results(
        messages,
        _collect_approval_responses(messages),
        [
            Content.from_function_result(call_id=None, result="ignored result"),
            Content.from_function_result(call_id="call_1", result="first result"),
        ],
    )

    assert len(messages) == 2
    assert messages[0].contents == [call_one]
    assert [(content.call_id, content.result) for content in messages[1].contents] == [("call_1", "first result")]


def test_replace_approval_contents_with_results_prunes_emptied_messages() -> None:
    """Messages whose contents are fully consumed during the first pass should be removed.

    When approval responses are paired with placeholder results, the responses are marked
    for removal in the first pass. If a message contained only such responses, it ends up
    with an empty `contents` list and the second pass should drop it from `messages`.
    """
    from agent_framework._tools import _collect_approval_responses, _replace_approval_contents_with_results

    call_one, request_one, response_one = _build_approved_tool_roundtrip(
        call_id="call_1", approval_id="approval_1", tool_name="first_tool"
    )
    call_two, request_two, response_two = _build_approved_tool_roundtrip(
        call_id="call_2", approval_id="approval_2", tool_name="second_tool"
    )

    messages = [
        Message(role="assistant", contents=[call_one, request_one, call_two, request_two]),
        Message(
            role="tool",
            contents=[
                Content.from_function_result(call_id="call_1", result="[APPROVAL_PENDING] first placeholder"),
                Content.from_function_result(call_id="call_2", result="[APPROVAL_PENDING] second placeholder"),
            ],
        ),
        # This user message holds only approval_responses whose placeholders are replaced
        # in the tool message above, so every content here is marked for removal and the
        # message itself becomes empty -> it must be pruned by the second pass.
        Message(role="user", contents=[response_one, response_two]),
    ]

    _replace_approval_contents_with_results(
        messages,
        _collect_approval_responses(messages),
        [
            Content.from_function_result(call_id="call_1", result="first result"),
            Content.from_function_result(call_id="call_2", result="second result"),
        ],
    )

    # The now-empty user message should have been pruned, leaving just the assistant
    # message and the tool message with the resolved results.
    assert len(messages) == 2
    assert messages[0].role == "assistant"
    assert messages[0].contents == [call_one, call_two]
    assert messages[1].role == "tool"
    assert [(content.call_id, content.result) for content in messages[1].contents] == [
        ("call_1", "first result"),
        ("call_2", "second result"),
    ]
    # Sanity-check: no leftover empty messages.
    assert all(msg.contents for msg in messages)


async def test_mixed_local_and_hosted_approval_flow(chat_client_base: SupportsChatGetResponse):
    """Test that mixed local + hosted MCP approvals are handled correctly.

    When a response contains both a local tool approval and a hosted MCP approval,
    the local approval should be processed normally while the hosted MCP approval
    should pass through untouched to the API.
    """

    @tool(name="local_function", approval_mode="always_require")
    def local_func(arg1: str) -> str:
        return f"Local {arg1}"

    # Simulate the LLM returning both a local function call and an MCP approval request
    local_fc = Content.from_function_call(call_id="call_local", name="local_function", arguments='{"arg1": "test"}')
    mcp_fc = Content.from_function_call(
        call_id="mcpr_hosted",
        name="microsoft_docs_search",
        arguments='{"query": "azure"}',
        additional_properties={"server_label": "Microsoft_Learn_MCP"},
    )
    mcp_approval_request = Content.from_function_approval_request(id="mcpr_hosted", function_call=mcp_fc)

    # First response: LLM returns a local function call that needs approval
    chat_client_base.run_responses = [
        ChatResponse(messages=Message(role="assistant", contents=[local_fc])),
        # After local approval + hosted approval, the final response
        ChatResponse(messages=Message(role="assistant", contents=["Done with both tools."])),
    ]

    # User approves the local function call
    local_approval_response = Content.from_function_approval_response(
        approved=True, id="call_local", function_call=local_fc
    )
    # User also has an MCP approval response (hosted)
    mcp_approval_response = mcp_approval_request.to_function_approval_response(approved=True)

    messages = [
        Message(role="user", contents=["Search docs and run local"]),
        Message(role="assistant", contents=[local_fc, mcp_approval_request]),
        Message(role="user", contents=[local_approval_response]),
        Message(role="user", contents=[mcp_approval_response]),
    ]

    response = await chat_client_base.get_response(
        messages,
        options={"tool_choice": "auto", "tools": [local_func]},
    )

    assert response is not None
    # The hosted MCP approval contents should NOT have been mutated
    assistant_msg = messages[1]
    assert assistant_msg.contents[1].type == "function_approval_request"
    mcp_user_msg = messages[3]
    assert mcp_user_msg.contents[0].type == "function_approval_response"


async def test_unapproved_tool_execution_raises_exception(chat_client_base: SupportsChatGetResponse):
    """Test that attempting to execute an unapproved tool raises ToolException."""

    @tool(name="test_function", approval_mode="always_require")
    def test_func(arg1: str) -> str:
        return f"Result {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="test_function", arguments='{"arg1": "value1"}'),
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    # Get approval request
    response1 = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [test_func]}
    )

    approval_req = [c for c in response1.messages[0].contents if c.type == "function_approval_request"][0]

    # Create a rejection response (approved=False)
    rejection_response = Content.from_function_approval_response(
        id=approval_req.id,
        function_call=approval_req.function_call,
        approved=False,
    )

    # Continue conversation with rejection
    all_messages = response1.messages + [Message(role="user", contents=[rejection_response])]

    # This should handle the rejection gracefully (not raise ToolException to user)
    await chat_client_base.get_response(all_messages, options={"tool_choice": "auto", "tools": [test_func]})

    # Should have a rejection result
    rejection_result = next(
        (
            content
            for msg in all_messages
            for content in msg.contents
            if content.type == "function_result" and "rejected" in (content.result or content.exception or "").lower()
        ),
        None,
    )
    assert rejection_result is not None


async def test_approved_function_call_with_error_without_detailed_errors(chat_client_base: SupportsChatGetResponse):
    """Test that approved functions that raise errors return generic error messages.

    When include_detailed_errors=False.
    """

    exec_counter = 0

    @tool(name="error_func", approval_mode="always_require")
    def error_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        raise ValueError("Specific error from approved function")

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[Content.from_function_call(call_id="1", name="error_func", arguments='{"arg1": "value1"}')],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    # Set include_detailed_errors to False (default)
    chat_client_base.function_invocation_configuration["include_detailed_errors"] = False

    # Get approval request
    response1 = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [error_func]}
    )

    approval_req = [c for c in response1.messages[0].contents if c.type == "function_approval_request"][0]

    # Approve the function
    approval_response = Content.from_function_approval_response(
        id=approval_req.id,
        function_call=approval_req.function_call,
        approved=True,
    )

    all_messages = response1.messages + [Message(role="user", contents=[approval_response])]

    # Execute the approved function (which will error)
    await chat_client_base.get_response(all_messages, options={"tool_choice": "auto", "tools": [error_func]})

    # Should have executed the function
    assert exec_counter == 1

    # Should have an error result with generic message
    error_result = next(
        (
            content
            for msg in all_messages
            for content in msg.contents
            if content.type == "function_result" and content.exception is not None
        ),
        None,
    )
    assert error_result is not None
    assert error_result.result is not None
    assert "Error: Function failed." in error_result.result
    assert "Specific error from approved function" not in error_result.result  # Detail not included


async def test_approved_function_call_with_error_with_detailed_errors(chat_client_base: SupportsChatGetResponse):
    """Test that approved functions that raise errors return detailed error messages.

    When include_detailed_errors=True.
    """

    exec_counter = 0

    @tool(name="error_func", approval_mode="always_require")
    def error_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        raise ValueError("Specific error from approved function")

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[Content.from_function_call(call_id="1", name="error_func", arguments='{"arg1": "value1"}')],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    # Set include_detailed_errors to True
    chat_client_base.function_invocation_configuration["include_detailed_errors"] = True

    # Get approval request
    response1 = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [error_func]}
    )

    approval_req = [c for c in response1.messages[0].contents if c.type == "function_approval_request"][0]

    # Approve the function
    approval_response = Content.from_function_approval_response(
        id=approval_req.id,
        function_call=approval_req.function_call,
        approved=True,
    )

    all_messages = response1.messages + [Message(role="user", contents=[approval_response])]

    # Execute the approved function (which will error)
    await chat_client_base.get_response(all_messages, options={"tool_choice": "auto", "tools": [error_func]})

    # Should have executed the function
    assert exec_counter == 1

    # Should have an error result with detailed message
    error_result = next(
        (
            content
            for msg in all_messages
            for content in msg.contents
            if content.type == "function_result" and content.exception is not None
        ),
        None,
    )
    assert error_result is not None
    assert error_result.result is not None
    assert "Error: Function failed." in error_result.result
    assert "Exception:" in error_result.result
    assert "Specific error from approved function" in error_result.result  # Detail included


async def test_approved_function_call_with_validation_error(chat_client_base: SupportsChatGetResponse):
    """Test that approved functions with validation errors are handled correctly."""

    exec_counter = 0

    @tool(name="typed_func", approval_mode="always_require")
    def typed_func(arg1: int) -> str:  # Expects int, not str
        nonlocal exec_counter
        exec_counter += 1
        return f"Got {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="typed_func", arguments='{"arg1": "not_an_int"}')
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    # Set include_detailed_errors to True to see validation details
    chat_client_base.function_invocation_configuration["include_detailed_errors"] = True

    # Get approval request
    response1 = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [typed_func]}
    )

    approval_req = [c for c in response1.messages[0].contents if c.type == "function_approval_request"][0]

    # Approve the function (even though it will fail validation)
    approval_response = Content.from_function_approval_response(
        id=approval_req.id,
        function_call=approval_req.function_call,
        approved=True,
    )

    all_messages = response1.messages + [Message(role="user", contents=[approval_response])]

    # Execute the approved function (which will fail validation)
    await chat_client_base.get_response(all_messages, options={"tool_choice": "auto", "tools": [typed_func]})

    # Should NOT have executed the function (validation failed before execution)
    assert exec_counter == 0

    # Should have a validation error result
    error_result = next(
        (
            content
            for msg in all_messages
            for content in msg.contents
            if content.type == "function_result" and content.exception is not None
        ),
        None,
    )
    assert error_result is not None
    assert error_result.result is not None
    assert "Argument parsing failed" in error_result.result


async def test_approved_function_call_successful_execution(chat_client_base: SupportsChatGetResponse):
    """Test that approved functions execute successfully when no errors occur."""

    exec_counter = 0

    @tool(name="success_func", approval_mode="always_require")
    def success_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Success {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[Content.from_function_call(call_id="1", name="success_func", arguments='{"arg1": "value1"}')],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    # Get approval request
    response1 = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [success_func]}
    )

    approval_req = [c for c in response1.messages[0].contents if c.type == "function_approval_request"][0]

    # Approve the function
    approval_response = Content.from_function_approval_response(
        id=approval_req.id,
        function_call=approval_req.function_call,
        approved=True,
    )

    all_messages = response1.messages + [Message(role="user", contents=[approval_response])]

    # Execute the approved function
    await chat_client_base.get_response(all_messages, options={"tool_choice": "auto", "tools": [success_func]})

    # Should have executed successfully
    assert exec_counter == 1

    # Should have a success result
    success_result = next(
        (
            content
            for msg in all_messages
            for content in msg.contents
            if content.type == "function_result" and content.exception is None
        ),
        None,
    )
    assert success_result is not None
    assert success_result.result == "Success value1"


async def test_declaration_only_tool(chat_client_base: SupportsChatGetResponse):
    """Test that declaration_only tools without implementation (func=None) are not executed."""
    from agent_framework import FunctionTool

    # Create a truly declaration-only function with no implementation
    declaration_func = FunctionTool(
        name="declaration_func",
        func=None,
        description="A declaration-only function for testing",
        input_model={"type": "object", "properties": {"arg1": {"type": "string"}}, "required": ["arg1"]},
    )

    # Verify it's marked as declaration_only
    assert declaration_func.declaration_only is True

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="declaration_func", arguments='{"arg1": "value1"}')
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    response = await chat_client_base.get_response(
        "hello", options={"tool_choice": "auto", "tools": [declaration_func]}
    )

    # Should have the function call in messages but not a result
    function_calls = [
        content
        for msg in response.messages
        for content in msg.contents
        if content.type == "function_call" and content.name == "declaration_func"
    ]
    assert len(function_calls) >= 1

    # Should not have a function result
    function_results = [
        content
        for msg in response.messages
        for content in msg.contents
        if content.type == "function_result" and content.call_id == "1"
    ]
    assert len(function_results) == 0


async def test_multiple_function_calls_parallel_execution(chat_client_base: SupportsChatGetResponse):
    """Test that multiple function calls are executed in parallel."""
    import asyncio

    exec_order = []

    @tool(name="func1", approval_mode="never_require")
    async def func1(arg1: str) -> str:
        exec_order.append("func1_start")
        await asyncio.sleep(0.01)  # Small delay
        exec_order.append("func1_end")
        return f"Result1 {arg1}"

    @tool(name="func2", approval_mode="never_require")
    async def func2(arg1: str) -> str:
        exec_order.append("func2_start")
        await asyncio.sleep(0.01)  # Small delay
        exec_order.append("func2_end")
        return f"Result2 {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="func1", arguments='{"arg1": "value1"}'),
                    Content.from_function_call(call_id="2", name="func2", arguments='{"arg1": "value2"}'),
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [func1, func2]}
    )

    # Both functions should have been executed
    assert "func1_start" in exec_order
    assert "func1_end" in exec_order
    assert "func2_start" in exec_order
    assert "func2_end" in exec_order

    # Should have results for both
    results = [content for msg in response.messages for content in msg.contents if content.type == "function_result"]
    assert len(results) == 2


async def test_callable_function_converted_to_tool(chat_client_base: SupportsChatGetResponse):
    """Test that plain callable functions are converted to FunctionTool."""
    exec_counter = 0

    @tool(approval_mode="never_require")
    def plain_function(arg1: str) -> str:
        """A plain function without decorator."""
        nonlocal exec_counter
        exec_counter += 1
        return f"Plain {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="plain_function", arguments='{"arg1": "value1"}')
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    # Pass plain function (will be auto-converted)
    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [plain_function]}
    )

    # Function should be executed
    assert exec_counter == 1
    result = next(content for msg in response.messages for content in msg.contents if content.type == "function_result")
    assert result.result == "Plain value1"


async def test_conversation_id_handling(chat_client_base: SupportsChatGetResponse):
    """Test that conversation_id is properly handled and messages are cleared."""

    @tool(name="test_function", approval_mode="never_require")
    def test_func(arg1: str) -> str:
        return f"Result {arg1}"

    # Return a response with a conversation_id
    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="test_function", arguments='{"arg1": "value1"}')
                ],
            ),
            conversation_id="conv_123",  # Simulate service-side thread
        ),
        ChatResponse(
            messages=Message(role="assistant", contents=["done"]),
            conversation_id="conv_123",
        ),
    ]

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [test_func]}
    )

    # Should have executed the function
    results = [content for msg in response.messages for content in msg.contents if content.type == "function_result"]
    assert len(results) >= 1
    assert response.conversation_id == "conv_123"


async def test_function_result_appended_to_existing_assistant_message(chat_client_base: SupportsChatGetResponse):
    """Test that function results are appended to existing assistant message when appropriate."""

    @tool(name="test_function", approval_mode="never_require")
    def test_func(arg1: str) -> str:
        return f"Result {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="test_function", arguments='{"arg1": "value1"}')
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [test_func]}
    )

    # Should have messages with both function call and function result
    assert len(response.messages) >= 2
    # Check that we have both a function call and a function result
    has_call = any(content.type == "function_call" for msg in response.messages for content in msg.contents)
    has_result = any(content.type == "function_result" for msg in response.messages for content in msg.contents)
    assert has_call
    assert has_result


@pytest.mark.parametrize("max_iterations", [3])
async def test_error_recovery_resets_counter(chat_client_base: SupportsChatGetResponse):
    """Test that error counter resets after a successful function call."""

    call_count = 0

    @tool(name="sometimes_fails", approval_mode="never_require")
    def sometimes_fails(arg1: str) -> str:
        nonlocal call_count
        call_count += 1
        if call_count == 1:
            raise ValueError("First call fails")
        return f"Success {arg1}"

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="sometimes_fails", arguments='{"arg1": "value1"}')
                ],
            )
        ),
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="2", name="sometimes_fails", arguments='{"arg1": "value2"}')
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [sometimes_fails]}
    )

    # Should have both an error and a success
    error_results = [
        content
        for msg in response.messages
        for content in msg.contents
        if content.type == "function_result" and content.exception
    ]
    success_results = [
        content
        for msg in response.messages
        for content in msg.contents
        if content.type == "function_result" and not content.exception
    ]

    assert len(error_results) >= 1
    assert len(success_results) >= 1
    assert call_count == 2  # Both calls executed


# ==================== STREAMING SCENARIO TESTS ====================


async def test_streaming_approval_request_generated(chat_client_base: SupportsChatGetResponse):
    """Test that approval requests are generated correctly in streaming mode."""
    exec_counter = 0

    @tool(name="test_func", approval_mode="always_require")
    def func_with_approval(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Result {arg1}"

    # Setup: function call that requires approval, streamed
    chat_client_base.streaming_responses = [
        [
            ChatResponseUpdate(
                contents=[Content.from_function_call(call_id="1", name="test_func", arguments='{"arg1": "value1"}')],
                role="assistant",
            ),
        ],
    ]

    # Get the streaming response with approval request
    updates = []
    async for update in chat_client_base.get_response(
        "hello", options={"tool_choice": "auto", "tools": [func_with_approval]}, stream=True
    ):
        updates.append(update)

    # Should have function call update and approval request
    approval_requests = [
        content for update in updates for content in update.contents if content.type == "function_approval_request"
    ]
    assert len(approval_requests) == 1
    assert approval_requests[0].function_call.name == "test_func"
    assert exec_counter == 0  # Function not executed yet due to approval requirement


async def test_streaming_max_iterations_limit(chat_client_base: SupportsChatGetResponse):
    """Test that MAX_ITERATIONS in streaming mode limits function call loops."""
    exec_counter = 0

    @tool(name="test_function", approval_mode="never_require")
    def ai_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Processed {arg1}"

    # Set up multiple function call responses to create a loop
    chat_client_base.streaming_responses = [
        [
            ChatResponseUpdate(
                contents=[Content.from_function_call(call_id="1", name="test_function", arguments='{"arg1":')],
                role="assistant",
            ),
            ChatResponseUpdate(
                contents=[Content.from_function_call(call_id="1", name="test_function", arguments='"value1"}')],
                role="assistant",
            ),
        ],
        [
            ChatResponseUpdate(
                contents=[Content.from_function_call(call_id="2", name="test_function", arguments='{"arg1":')],
                role="assistant",
            ),
            ChatResponseUpdate(
                contents=[Content.from_function_call(call_id="2", name="test_function", arguments='"value2"}')],
                role="assistant",
            ),
        ],
        # Failsafe response when tool_choice is set to "none"
        [ChatResponseUpdate(contents=[Content.from_text(text="giving up on tools")], role="assistant")],
    ]

    # Set max_iterations to 1 in additional_properties
    chat_client_base.function_invocation_configuration["max_iterations"] = 1

    updates = []
    async for update in chat_client_base.get_response(
        "hello", options={"tool_choice": "auto", "tools": [ai_func]}, stream=True
    ):
        updates.append(update)

    # With max_iterations=1, we should only execute first function
    assert exec_counter == 1  # Only first function executed
    # Should have the failsafe message
    last_text = "".join(u.text or "" for u in updates if u.text)
    assert "I broke out of the function invocation loop..." in last_text


async def test_streaming_function_invocation_config_enabled_false(chat_client_base: SupportsChatGetResponse):
    """Test that setting enabled=False disables function invocation in streaming mode."""
    exec_counter = 0

    @tool(name="test_function", approval_mode="never_require")
    def ai_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Processed {arg1}"

    chat_client_base.streaming_responses = [
        [ChatResponseUpdate(contents=[Content.from_text(text="response without function calling")], role="assistant")],
    ]

    # Disable function invocation
    chat_client_base.function_invocation_configuration["enabled"] = False

    updates = []
    async for update in chat_client_base.get_response(
        "hello", options={"tool_choice": "auto", "tools": [ai_func]}, stream=True
    ):
        updates.append(update)

    # Function should not be executed - when enabled=False, the loop doesn't run
    assert exec_counter == 0
    # The response should be from the mock client
    assert len(updates) > 0


async def test_streaming_function_invocation_config_max_consecutive_errors(chat_client_base: SupportsChatGetResponse):
    """Test that max_consecutive_errors_per_request limits error retries in streaming mode."""

    @tool(name="error_function", approval_mode="never_require")
    def error_func(arg1: str) -> str:
        raise ValueError("Function error")

    # Set up multiple function call responses that will all error
    chat_client_base.streaming_responses = [
        [
            ChatResponseUpdate(
                contents=[
                    Content.from_function_call(call_id="1", name="error_function", arguments='{"arg1": "value1"}')
                ],
                role="assistant",
            ),
        ],
        [
            ChatResponseUpdate(
                contents=[
                    Content.from_function_call(call_id="2", name="error_function", arguments='{"arg1": "value2"}')
                ],
                role="assistant",
            ),
        ],
        [
            ChatResponseUpdate(
                contents=[
                    Content.from_function_call(call_id="3", name="error_function", arguments='{"arg1": "value3"}')
                ],
                role="assistant",
            ),
        ],
        [ChatResponseUpdate(contents=[Content.from_text(text="final response")], role="assistant")],
    ]

    # Set max_consecutive_errors to 2
    chat_client_base.function_invocation_configuration["max_consecutive_errors_per_request"] = 2

    updates = []
    async for update in chat_client_base.get_response(
        "hello", options={"tool_choice": "auto", "tools": [error_func]}, stream=True
    ):
        updates.append(update)

    # Should stop after 2 consecutive errors
    error_results = [
        content
        for update in updates
        for content in update.contents
        if content.type == "function_result" and content.exception
    ]
    # At least one error occurred
    assert len(error_results) >= 1
    # Should have stopped making new function calls after hitting the error limit
    function_calls = [content for update in updates for content in update.contents if content.type == "function_call"]
    # Should have made at most 2 function calls before stopping
    assert len(function_calls) <= 2


async def test_streaming_function_invocation_stop_clears_conversation_id(chat_client_base: SupportsChatGetResponse):
    """Streaming stop-path responses should not carry a continuation conversation_id."""

    @tool(name="error_function", approval_mode="never_require")
    def error_func(arg1: str) -> str:
        raise ValueError("Function error")

    chat_client_base.streaming_responses = [
        [
            ChatResponseUpdate(
                contents=[
                    Content.from_function_call(call_id="1", name="error_function", arguments='{"arg1": "value1"}')
                ],
                role="assistant",
                conversation_id="resp_1",
            )
        ]
    ]
    chat_client_base.function_invocation_configuration["max_consecutive_errors_per_request"] = 1
    session_stub = type("SessionStub", (), {"service_session_id": "resp_seed"})()

    stream = chat_client_base.get_response(
        "hello",
        options={"tool_choice": "auto", "tools": [error_func]},
        stream=True,
        client_kwargs={"session": session_stub},
    )
    async for _ in stream:
        pass
    response = await stream.get_final_response()

    # After the stop-path cleanup call, the accumulated stream response keeps the
    # conversation_id from the first inner call; the cleanup call's own response id
    # is what matters for server-side resolution but is not reflected in the mock here.
    assert response is not None


async def test_streaming_function_invocation_config_terminate_on_unknown_calls_false(
    chat_client_base: SupportsChatGetResponse,
):
    """Test that terminate_on_unknown_calls=False returns error message for unknown functions in streaming mode."""
    exec_counter = 0

    @tool(name="known_function", approval_mode="never_require")
    def known_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Processed {arg1}"

    chat_client_base.streaming_responses = [
        [
            ChatResponseUpdate(
                contents=[
                    Content.from_function_call(call_id="1", name="unknown_function", arguments='{"arg1": "value1"}')
                ],
                role="assistant",
            ),
        ],
        [ChatResponseUpdate(contents=[Content.from_text(text="done")], role="assistant")],
    ]

    # Set terminate_on_unknown_calls to False (default)
    chat_client_base.function_invocation_configuration["terminate_on_unknown_calls"] = False

    updates = []
    async for update in chat_client_base.get_response(
        "hello", options={"tool_choice": "auto", "tools": [known_func]}, stream=True
    ):
        updates.append(update)

    # Should have a result message indicating the tool wasn't found
    result_contents = [
        content for update in updates for content in update.contents if content.type == "function_result"
    ]
    assert len(result_contents) >= 1
    result_str = result_contents[0].result or result_contents[0].exception or ""
    assert "not found" in result_str.lower()
    assert exec_counter == 0  # Known function not executed


@pytest.mark.skip(reason="Failsafe behavior needs investigation in unified API")
async def test_streaming_function_invocation_config_terminate_on_unknown_calls_true(
    chat_client_base: SupportsChatGetResponse,
):
    """Test that terminate_on_unknown_calls=True stops execution on unknown functions in streaming mode."""
    exec_counter = 0

    @tool(name="known_function", approval_mode="never_require")
    def known_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Processed {arg1}"

    chat_client_base.streaming_responses = [
        [
            ChatResponseUpdate(
                contents=[
                    Content.from_function_call(call_id="1", name="unknown_function", arguments='{"arg1": "value1"}')
                ],
                role="assistant",
            ),
        ],
    ]

    # Set terminate_on_unknown_calls to True
    chat_client_base.function_invocation_configuration["terminate_on_unknown_calls"] = True

    # Should raise an exception when encountering an unknown function
    with pytest.raises(KeyError, match='Error: Requested function "unknown_function" not found'):
        async for _ in chat_client_base.get_response(
            [Message(role="user", contents=["hello"])], options={"tool_choice": "auto", "tools": [known_func]}
        ):
            pass

    assert exec_counter == 0


async def test_streaming_function_invocation_config_include_detailed_errors_true(
    chat_client_base: SupportsChatGetResponse,
):
    """Test that include_detailed_errors=True returns detailed error information in streaming mode."""

    @tool(name="error_function", approval_mode="never_require")
    def error_func(arg1: str) -> str:
        raise ValueError("Specific error message that should appear")

    chat_client_base.streaming_responses = [
        [
            ChatResponseUpdate(
                contents=[
                    Content.from_function_call(call_id="1", name="error_function", arguments='{"arg1": "value1"}')
                ],
                role="assistant",
            ),
        ],
        [ChatResponseUpdate(contents=[Content.from_text(text="done")], role="assistant")],
    ]

    # Set include_detailed_errors to True
    chat_client_base.function_invocation_configuration["include_detailed_errors"] = True

    updates = []
    async for update in chat_client_base.get_response(
        "hello", options={"tool_choice": "auto", "tools": [error_func]}, stream=True
    ):
        updates.append(update)

    # Should have detailed error message
    error_result = next(
        content for update in updates for content in update.contents if content.type == "function_result"
    )
    assert error_result.result is not None
    assert error_result.exception is not None
    assert "Specific error message that should appear" in error_result.result
    assert "Exception:" in error_result.result


async def test_streaming_function_invocation_config_include_detailed_errors_false(
    chat_client_base: SupportsChatGetResponse,
):
    """Test that include_detailed_errors=False returns generic error messages in streaming mode."""

    @tool(name="error_function", approval_mode="never_require")
    def error_func(arg1: str) -> str:
        raise ValueError("Specific error message that should not appear")

    chat_client_base.streaming_responses = [
        [
            ChatResponseUpdate(
                contents=[
                    Content.from_function_call(call_id="1", name="error_function", arguments='{"arg1": "value1"}')
                ],
                role="assistant",
            ),
        ],
        [ChatResponseUpdate(contents=[Content.from_text(text="done")], role="assistant")],
    ]

    # Set include_detailed_errors to False (default)
    chat_client_base.function_invocation_configuration["include_detailed_errors"] = False

    updates = []
    async for update in chat_client_base.get_response(
        "hello", options={"tool_choice": "auto", "tools": [error_func]}, stream=True
    ):
        updates.append(update)

    # Should have a generic error message
    error_result = next(
        content for update in updates for content in update.contents if content.type == "function_result"
    )
    assert error_result.result is not None
    assert error_result.exception is not None
    assert "Specific error message" not in error_result.result
    assert "Error:" in error_result.result  # Generic error prefix


async def test_streaming_argument_validation_error_with_detailed_errors(chat_client_base: SupportsChatGetResponse):
    """Test that argument validation errors include details when include_detailed_errors=True in streaming mode."""

    @tool(name="typed_function", approval_mode="never_require")
    def typed_func(arg1: int) -> str:  # Expects int, not str
        return f"Got {arg1}"

    chat_client_base.streaming_responses = [
        [
            ChatResponseUpdate(
                contents=[
                    Content.from_function_call(call_id="1", name="typed_function", arguments='{"arg1": "not_an_int"}')
                ],
                role="assistant",
            ),
        ],
        [ChatResponseUpdate(contents=[Content.from_text(text="done")], role="assistant")],
    ]

    # Set include_detailed_errors to True
    chat_client_base.function_invocation_configuration["include_detailed_errors"] = True

    updates = []
    async for update in chat_client_base.get_response(
        "hello", options={"tool_choice": "auto", "tools": [typed_func]}, stream=True
    ):
        updates.append(update)

    # Should have detailed validation error
    error_result = next(
        content for update in updates for content in update.contents if content.type == "function_result"
    )
    assert error_result.result is not None
    assert error_result.exception is not None
    assert "Argument parsing failed" in error_result.result
    assert "Exception:" in error_result.result  # Detailed error included


async def test_streaming_argument_validation_error_without_detailed_errors(chat_client_base: SupportsChatGetResponse):
    """Test that argument validation errors are generic when include_detailed_errors=False in streaming mode."""

    @tool(name="typed_function", approval_mode="never_require")
    def typed_func(arg1: int) -> str:  # Expects int, not str
        return f"Got {arg1}"

    chat_client_base.streaming_responses = [
        [
            ChatResponseUpdate(
                contents=[
                    Content.from_function_call(call_id="1", name="typed_function", arguments='{"arg1": "not_an_int"}')
                ],
                role="assistant",
            ),
        ],
        [ChatResponseUpdate(contents=[Content.from_text(text="done")], role="assistant")],
    ]

    # Set include_detailed_errors to False (default)
    chat_client_base.function_invocation_configuration["include_detailed_errors"] = False

    updates = []
    async for update in chat_client_base.get_response(
        "hello", options={"tool_choice": "auto", "tools": [typed_func]}, stream=True
    ):
        updates.append(update)

    # Should have generic validation error
    error_result = next(
        content for update in updates for content in update.contents if content.type == "function_result"
    )
    assert error_result.result is not None
    assert error_result.exception is not None
    assert "Argument parsing failed" in error_result.result
    assert "Exception:" not in error_result.result  # No detailed error


async def test_streaming_multiple_function_calls_parallel_execution(chat_client_base: SupportsChatGetResponse):
    """Test that multiple function calls are executed in parallel in streaming mode."""

    exec_order = []

    @tool(name="func1", approval_mode="never_require")
    async def func1(arg1: str) -> str:
        exec_order.append("func1_start")
        await asyncio.sleep(0.01)  # Small delay
        exec_order.append("func1_end")
        return f"Result1 {arg1}"

    @tool(name="func2", approval_mode="never_require")
    async def func2(arg1: str) -> str:
        exec_order.append("func2_start")
        await asyncio.sleep(0.01)  # Small delay
        exec_order.append("func2_end")
        return f"Result2 {arg1}"

    chat_client_base.streaming_responses = [
        [
            ChatResponseUpdate(
                contents=[Content.from_function_call(call_id="1", name="func1", arguments='{"arg1": "value1"}')],
                role="assistant",
            ),
            ChatResponseUpdate(
                contents=[Content.from_function_call(call_id="2", name="func2", arguments='{"arg1": "value2"}')],
                role="assistant",
            ),
        ],
        [ChatResponseUpdate(contents=[Content.from_text(text="done")], role="assistant")],
    ]

    updates = []
    async for update in chat_client_base.get_response(
        "hello", options={"tool_choice": "auto", "tools": [func1, func2]}, stream=True
    ):
        updates.append(update)

    # Both functions should have been executed
    assert "func1_start" in exec_order
    assert "func1_end" in exec_order
    assert "func2_start" in exec_order
    assert "func2_end" in exec_order

    # Should have results for both
    results = [content for update in updates for content in update.contents if content.type == "function_result"]
    assert len(results) == 2


async def test_streaming_approval_requests_in_assistant_message(chat_client_base: SupportsChatGetResponse):
    """Approval requests should be added to assistant updates in streaming mode."""
    exec_counter = 0

    @tool(name="test_func", approval_mode="always_require")
    def func_with_approval(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Result {arg1}"

    chat_client_base.streaming_responses = [
        [
            ChatResponseUpdate(
                contents=[
                    Content.from_function_call(call_id="1", name="test_func", arguments='{"arg1": "value1"}'),
                ],
                role="assistant",
            ),
        ],
    ]

    updates = []
    async for update in chat_client_base.get_response(
        "hello", options={"tool_choice": "auto", "tools": [func_with_approval]}, stream=True
    ):
        updates.append(update)

    # Should have updates containing both the call and approval request
    approval_requests = [
        content for update in updates for content in update.contents if content.type == "function_approval_request"
    ]
    assert len(approval_requests) == 1
    assert exec_counter == 0


async def test_streaming_error_recovery_resets_counter(chat_client_base: SupportsChatGetResponse):
    """Test that error counter resets after a successful function call in streaming mode."""

    call_count = 0

    @tool(name="sometimes_fails", approval_mode="never_require")
    def sometimes_fails(arg1: str) -> str:
        nonlocal call_count
        call_count += 1
        if call_count == 1:
            raise ValueError("First call fails")
        return f"Success {arg1}"

    chat_client_base.streaming_responses = [
        [
            ChatResponseUpdate(
                contents=[
                    Content.from_function_call(call_id="1", name="sometimes_fails", arguments='{"arg1": "value1"}')
                ],
                role="assistant",
            ),
        ],
        [
            ChatResponseUpdate(
                contents=[
                    Content.from_function_call(call_id="2", name="sometimes_fails", arguments='{"arg1": "value2"}')
                ],
                role="assistant",
            ),
        ],
        [ChatResponseUpdate(contents=[Content.from_text(text="done")], role="assistant")],
    ]

    updates = []
    async for update in chat_client_base.get_response(
        "hello", options={"tool_choice": "auto", "tools": [sometimes_fails]}, stream=True
    ):
        updates.append(update)

    # Should have both an error and a success
    error_results = [
        content
        for update in updates
        for content in update.contents
        if content.type == "function_result" and content.exception
    ]
    success_results = [
        content
        for update in updates
        for content in update.contents
        if content.type == "function_result" and content.result
    ]

    assert len(error_results) >= 1
    assert len(success_results) >= 1
    assert call_count == 2  # Both calls executed


class TerminateLoopMiddleware(FunctionMiddleware):
    """Middleware that raises MiddlewareTermination to exit the function calling loop."""

    async def process(self, context: FunctionInvocationContext, next_handler: Callable[[], Awaitable[None]]) -> None:
        # Set result to a simple value - the framework will wrap it in FunctionResultContent
        context.result = "terminated by middleware"
        raise MiddlewareTermination


async def test_terminate_loop_single_function_call(chat_client_base: SupportsChatGetResponse):
    """Test that terminate_loop=True exits the function calling loop after single function call."""
    exec_counter = 0

    @tool(name="test_function", approval_mode="never_require")
    def ai_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Processed {arg1}"

    # Queue up two responses: function call, then final text
    # If terminate_loop works, only the first response should be consumed
    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="test_function", arguments='{"arg1": "value1"}')
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    response = await chat_client_base.get_response(
        "hello",
        options={"tool_choice": "auto", "tools": [ai_func]},
        client_kwargs={"middleware": [TerminateLoopMiddleware()]},
    )

    # Function should NOT have been executed - middleware intercepted it
    assert exec_counter == 0

    # There should be 2 messages: assistant with function call, tool result from middleware
    # The loop should NOT have continued to call the LLM again
    assert len(response.messages) == 2
    assert response.messages[0].role == "assistant"
    assert response.messages[0].contents[0].type == "function_call"
    assert response.messages[1].role == "tool"
    assert response.messages[1].contents[0].type == "function_result"
    assert response.messages[1].contents[0].result == "terminated by middleware"

    # Verify the second response is still in the queue (wasn't consumed)
    assert len(chat_client_base.run_responses) == 1


class SelectiveTerminateMiddleware(FunctionMiddleware):
    """Only terminates for terminating_function."""

    async def process(self, context: FunctionInvocationContext, next_handler: Callable[[], Awaitable[None]]) -> None:
        if context.function.name == "terminating_function":
            # Set result to a simple value - the framework will wrap it in FunctionResultContent
            context.result = "terminated by middleware"
            raise MiddlewareTermination
        await next_handler()


async def test_terminate_loop_multiple_function_calls_one_terminates(chat_client_base: SupportsChatGetResponse):
    """Test that any(terminate_loop=True) exits loop even with multiple function calls."""
    normal_call_count = 0
    terminating_call_count = 0

    @tool(name="normal_function", approval_mode="never_require")
    def normal_func(arg1: str) -> str:
        nonlocal normal_call_count
        normal_call_count += 1
        return f"Normal {arg1}"

    @tool(name="terminating_function", approval_mode="never_require")
    def terminating_func(arg1: str) -> str:
        nonlocal terminating_call_count
        terminating_call_count += 1
        return f"Terminating {arg1}"

    # Queue up two responses: parallel function calls, then final text
    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="normal_function", arguments='{"arg1": "value1"}'),
                    Content.from_function_call(
                        call_id="2", name="terminating_function", arguments='{"arg1": "value2"}'
                    ),
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["done"])),
    ]

    response = await chat_client_base.get_response(
        "hello",
        options={"tool_choice": "auto", "tools": [normal_func, terminating_func]},
        client_kwargs={"middleware": [SelectiveTerminateMiddleware()]},
    )

    # normal_function should have executed (middleware calls next_handler)
    # terminating_function should NOT have executed (middleware intercepts it)
    assert normal_call_count == 1
    assert terminating_call_count == 0

    # There should be 2 messages: assistant with function calls, tool results
    # The loop should NOT have continued to call the LLM again
    assert len(response.messages) == 2
    assert response.messages[0].role == "assistant"
    assert len(response.messages[0].contents) == 2
    assert response.messages[1].role == "tool"
    # Both function results should be present
    assert len(response.messages[1].contents) == 2

    # Verify the second response is still in the queue (wasn't consumed)
    assert len(chat_client_base.run_responses) == 1


async def test_terminate_loop_streaming_single_function_call(chat_client_base: SupportsChatGetResponse):
    """Test that terminate_loop=True exits the streaming function calling loop."""
    exec_counter = 0

    @tool(name="test_function", approval_mode="never_require")
    def ai_func(arg1: str) -> str:
        nonlocal exec_counter
        exec_counter += 1
        return f"Processed {arg1}"

    # Queue up two streaming responses
    chat_client_base.streaming_responses = [
        [
            ChatResponseUpdate(
                contents=[
                    Content.from_function_call(call_id="1", name="test_function", arguments='{"arg1": "value1"}')
                ],
                role="assistant",
            ),
        ],
        [
            ChatResponseUpdate(
                contents=[Content.from_text(text="done")],
                role="assistant",
            )
        ],
    ]

    updates = []
    async for update in chat_client_base.get_response(
        "hello",
        options={"tool_choice": "auto", "tools": [ai_func]},
        client_kwargs={"middleware": [TerminateLoopMiddleware()]},
        stream=True,
    ):
        updates.append(update)

    # Function should NOT have been executed - middleware intercepted it
    assert exec_counter == 0

    # Should have function call update and function result update
    # The loop should NOT have continued to call the LLM again
    assert len(updates) == 2

    # Verify the second streaming response is still in the queue (wasn't consumed)
    assert len(chat_client_base.streaming_responses) == 1


async def test_conversation_id_updated_in_options_between_tool_iterations():
    """Test that conversation_id is updated in options dict between tool invocation iterations.

    This regression test ensures that when a tool call returns a new conversation_id,
    subsequent API calls in the same function invocation loop use the updated conversation_id.
    Without this fix, the old conversation_id would be used, causing "No tool call found"
    errors when submitting tool results to APIs like OpenAI Responses.
    """
    from collections.abc import AsyncIterable, MutableSequence, Sequence
    from typing import Any
    from unittest.mock import patch

    from agent_framework import (
        BaseChatClient,
        ChatResponse,
        ChatResponseUpdate,
        Content,
        Message,
        ResponseStream,
        tool,
    )
    from agent_framework._middleware import ChatMiddlewareLayer
    from agent_framework._tools import FunctionInvocationLayer

    # Track the conversation_id passed to each call
    conversation_ids_received: list[str | None] = []

    class TrackingChatClient(
        FunctionInvocationLayer,
        ChatMiddlewareLayer,
        BaseChatClient,
    ):
        def __init__(self) -> None:
            super().__init__(middleware=[])
            self.run_responses: list[ChatResponse] = []
            self.streaming_responses: list[list[ChatResponseUpdate]] = []
            self.call_count: int = 0

        def _inner_get_response(
            self,
            *,
            messages: MutableSequence[Message],
            stream: bool,
            options: dict[str, Any],
            **kwargs: Any,
        ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]:
            # Track what conversation_id was passed
            conversation_ids_received.append(options.get("conversation_id"))

            if stream:
                return self._get_streaming_response(messages=messages, options=options, **kwargs)

            async def _get() -> ChatResponse:
                self.call_count += 1
                if not self.run_responses:
                    return ChatResponse(messages=Message(role="assistant", contents=["done"]))
                return self.run_responses.pop(0)

            return _get()

        def _get_streaming_response(
            self,
            *,
            messages: MutableSequence[Message],
            options: dict[str, Any],
            **kwargs: Any,
        ) -> ResponseStream[ChatResponseUpdate, ChatResponse]:
            async def _stream() -> AsyncIterable[ChatResponseUpdate]:
                self.call_count += 1
                if not self.streaming_responses:
                    yield ChatResponseUpdate(
                        contents=[Content.from_text("done")], role="assistant", finish_reason="stop"
                    )
                    return
                response = self.streaming_responses.pop(0)
                for update in response:
                    yield update

            def _finalize(updates: Sequence[ChatResponseUpdate]) -> ChatResponse:
                return ChatResponse.from_updates(updates)

            return ResponseStream(_stream(), finalizer=_finalize)

    @tool(name="test_func", approval_mode="never_require")
    def test_func(arg1: str) -> str:
        return f"Result {arg1}"

    # Test non-streaming: conversation_id should be updated after first response
    with patch("agent_framework._tools.DEFAULT_MAX_ITERATIONS", 5):
        client = TrackingChatClient()

    # First response returns a function call WITH a new conversation_id
    # Second response (after tool execution) should receive the updated conversation_id
    client.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[Content.from_function_call(call_id="call_1", name="test_func", arguments='{"arg1": "v1"}')],
            ),
            conversation_id="conv_after_first_call",
        ),
        ChatResponse(
            messages=Message(role="assistant", contents=["done"]),
            conversation_id="conv_after_second_call",
        ),
    ]

    # Start with initial conversation_id
    await client.get_response(
        "hello",
        options={"tool_choice": "auto", "tools": [test_func], "conversation_id": "conv_initial"},
    )

    assert client.call_count == 2
    # First call should receive the initial conversation_id
    assert conversation_ids_received[0] == "conv_initial"
    # Second call (after tool execution) MUST receive the updated conversation_id
    assert conversation_ids_received[1] == "conv_after_first_call", (
        "conversation_id should be updated in options after receiving new conversation_id from API"
    )

    # Test streaming version too
    conversation_ids_received.clear()

    with patch("agent_framework._tools.DEFAULT_MAX_ITERATIONS", 5):
        streaming_client = TrackingChatClient()

    streaming_client.streaming_responses = [
        [
            ChatResponseUpdate(
                contents=[Content.from_function_call(call_id="call_2", name="test_func", arguments='{"arg1": "v2"}')],
                role="assistant",
                conversation_id="stream_conv_after_first",
            ),
        ],
        [
            ChatResponseUpdate(contents=[Content.from_text("streaming done")], role="assistant", finish_reason="stop"),
        ],
    ]

    response_stream = streaming_client.get_response(
        "hello",
        stream=True,
        options={"tool_choice": "auto", "tools": [test_func], "conversation_id": "stream_conv_initial"},
    )
    updates = []
    async for update in response_stream:
        updates.append(update)

    assert streaming_client.call_count == 2
    # First call should receive the initial conversation_id
    assert conversation_ids_received[0] == "stream_conv_initial"
    # Second call (after tool execution) MUST receive the updated conversation_id
    assert conversation_ids_received[1] == "stream_conv_after_first", (
        "streaming: conversation_id should be updated in options after receiving new conversation_id from API"
    )


async def test_streaming_function_calling_response_includes_reasoning_and_tool_results(
    chat_client_base: SupportsChatGetResponse,
):
    """Test that the finalized streaming response includes reasoning, function_call,
    function_result, and final text in its messages.

    This is critical for workflow chaining: when one agent's response is passed as
    input to the next agent, the conversation must include all items (reasoning,
    function_call, function_call_output) so the API can validate the history.
    """

    @tool(name="search", approval_mode="never_require")
    def search_func(query: str) -> str:
        return f"Found results for {query}"

    chat_client_base.streaming_responses = [
        [
            # First response: reasoning + function_call
            ChatResponseUpdate(
                contents=[
                    Content.from_text_reasoning(
                        id="rs_test123",
                        text="Let me search for that",
                        additional_properties={"status": "completed"},
                    )
                ],
                role="assistant",
            ),
            ChatResponseUpdate(
                contents=[
                    Content.from_function_call(
                        call_id="call_1",
                        name="search",
                        arguments='{"query": "test"}',
                        additional_properties={"fc_id": "fc_test456"},
                    )
                ],
                role="assistant",
            ),
        ],
        [
            # Second response: final text
            ChatResponseUpdate(
                contents=[Content.from_text(text="Here are the results")],
                role="assistant",
            ),
        ],
    ]

    stream = chat_client_base.get_response(
        "search for test", options={"tool_choice": "auto", "tools": [search_func]}, stream=True
    )

    updates = []
    async for update in stream:
        updates.append(update)
    response = await stream.get_final_response()

    # Verify all content types are in the response messages
    all_content_types = [c.type for msg in response.messages for c in msg.contents]
    assert "text_reasoning" in all_content_types, "Reasoning must be preserved in response messages"
    assert "function_call" in all_content_types, "Function call must be preserved in response messages"
    assert "function_result" in all_content_types, "Function result must be in response messages for chaining"
    assert "text" in all_content_types, "Final text must be in response messages"

    # Verify reasoning has the id preserved
    reasoning_contents = [c for msg in response.messages for c in msg.contents if c.type == "text_reasoning"]
    assert len(reasoning_contents) >= 1
    assert reasoning_contents[0].id == "rs_test123"


# region _update_conversation_id unit tests


class TestUpdateConversationId:
    """Tests for _update_conversation_id handling dict chat_options."""

    def test_chat_options_as_dict(self):
        """When chat_options is a plain dict, conversation_id should be set via key access."""
        from agent_framework._tools import _update_conversation_id

        kwargs: dict[str, Any] = {"chat_options": {}}
        _update_conversation_id(kwargs, "conv_1")
        assert kwargs["chat_options"]["conversation_id"] == "conv_1"

    def test_chat_options_as_typed_dict(self):
        """When chat_options is a ChatOptions TypedDict, conversation_id should be set via key access."""
        from agent_framework import ChatOptions
        from agent_framework._tools import _update_conversation_id

        opts: ChatOptions = {"temperature": 0.5}
        kwargs: dict[str, Any] = {"chat_options": opts}
        _update_conversation_id(kwargs, "conv_2")
        assert kwargs["chat_options"]["conversation_id"] == "conv_2"

    def test_no_chat_options_falls_back_to_kwargs(self):
        """When chat_options is absent, conversation_id should be set directly on kwargs."""
        from agent_framework._tools import _update_conversation_id

        kwargs: dict[str, Any] = {}
        _update_conversation_id(kwargs, "conv_4")
        assert kwargs["conversation_id"] == "conv_4"

    def test_none_conversation_id_is_noop(self):
        """When conversation_id is None, kwargs should not be modified."""
        from agent_framework._tools import _update_conversation_id

        kwargs: dict[str, Any] = {"chat_options": {}}
        _update_conversation_id(kwargs, None)
        assert "conversation_id" not in kwargs["chat_options"]
        assert "conversation_id" not in kwargs

    def test_options_dict_also_updated(self):
        """The optional options dict should also receive conversation_id."""
        from agent_framework._tools import _update_conversation_id

        kwargs: dict[str, Any] = {"chat_options": {}}
        options: dict[str, Any] = {}
        _update_conversation_id(kwargs, "conv_5", options)
        assert kwargs["chat_options"]["conversation_id"] == "conv_5"
        assert options["conversation_id"] == "conv_5"

    def test_dict_overwrites_existing_conversation_id(self):
        """When a dict already has a conversation_id, it should be overwritten."""
        from agent_framework._tools import _update_conversation_id

        kwargs: dict[str, Any] = {"chat_options": {"conversation_id": "old_id"}}
        _update_conversation_id(kwargs, "new_id")
        assert kwargs["chat_options"]["conversation_id"] == "new_id"


# endregion
async def test_user_input_request_propagates_through_as_tool(chat_client_base: SupportsChatGetResponse):
    """Test that user_input_request content from a sub-agent wrapped as a tool propagates to the parent response."""
    from agent_framework.exceptions import UserInputRequiredException

    @tool(name="delegate_agent", approval_mode="never_require")
    def delegate_tool(task: str) -> str:
        del task
        raise UserInputRequiredException(
            contents=[
                Content.from_oauth_consent_request(
                    consent_link="https://login.microsoftonline.com/consent",
                )
            ]
        )

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="delegate_agent", arguments='{"task": "do it"}'),
                ],
            )
        )
    ]

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["delegate this"])],
        options={"tool_choice": "auto", "tools": [delegate_tool]},
    )

    user_requests = [
        content
        for msg in response.messages
        for content in msg.contents
        if isinstance(content, Content) and content.user_input_request
    ]
    assert len(user_requests) == 1
    assert user_requests[0].type == "oauth_consent_request"
    assert user_requests[0].consent_link == "https://login.microsoftonline.com/consent"
    assert user_requests[0].user_input_request is True


async def test_user_input_request_multiple_contents_propagate(chat_client_base: SupportsChatGetResponse):
    """Test that multiple user_input_request items in a single exception all propagate to the parent response."""
    from agent_framework.exceptions import UserInputRequiredException

    @tool(name="multi_request_tool", approval_mode="never_require")
    def multi_request(task: str) -> str:
        del task
        raise UserInputRequiredException(
            contents=[
                Content.from_oauth_consent_request(
                    consent_link="https://example.com/consent1",
                ),
                Content.from_oauth_consent_request(
                    consent_link="https://example.com/consent2",
                ),
                Content.from_oauth_consent_request(
                    consent_link="https://example.com/consent3",
                ),
            ]
        )

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="multi_request_tool", arguments='{"task": "do it"}'),
                ],
            )
        )
    ]

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["do something"])],
        options={"tool_choice": "auto", "tools": [multi_request]},
    )

    user_requests = [
        content
        for msg in response.messages
        for content in msg.contents
        if isinstance(content, Content) and content.user_input_request
    ]
    assert len(user_requests) == 3
    consent_links = {r.consent_link for r in user_requests}
    assert consent_links == {
        "https://example.com/consent1",
        "https://example.com/consent2",
        "https://example.com/consent3",
    }


async def test_user_input_request_empty_contents_returns_fallback(chat_client_base: SupportsChatGetResponse):
    """Test that UserInputRequiredException with empty contents produces a fallback function_result."""
    from agent_framework.exceptions import UserInputRequiredException

    @tool(name="empty_request_tool", approval_mode="never_require")
    def empty_request(task: str) -> str:
        del task
        raise UserInputRequiredException(contents=[])

    chat_client_base.run_responses = [
        ChatResponse(
            messages=Message(
                role="assistant",
                contents=[
                    Content.from_function_call(call_id="1", name="empty_request_tool", arguments='{"task": "do it"}'),
                ],
            )
        ),
        ChatResponse(messages=Message(role="assistant", contents=["handled"])),
    ]

    response = await chat_client_base.get_response(
        [Message(role="user", contents=["do something"])],
        options={"tool_choice": "auto", "tools": [empty_request]},
    )

    # With empty contents, the handler returns a function_result with an error message
    # and the loop continues to the next chat response.
    function_results = [
        content for msg in response.messages for content in msg.contents if content.type == "function_result"
    ]
    assert len(function_results) >= 1
    assert any("user input" in (fr.result or "").lower() for fr in function_results)


# region Progressive tool exposure (FunctionInvocationContext.add_tools / remove_tools)


def _pte_function_call_response(call_id: str, name: str, arguments: str = "{}") -> ChatResponse:
    return ChatResponse(
        messages=Message(
            role="assistant",
            contents=[Content.from_function_call(call_id=call_id, name=name, arguments=arguments)],
        )
    )


def _pte_text_response(text: str = "done") -> ChatResponse:
    return ChatResponse(messages=Message(role="assistant", contents=[text]))


@tool(name="factorial", approval_mode="never_require")
def _pte_factorial(n: int) -> int:
    """Compute the factorial of n."""
    result = 1
    for value in range(2, n + 1):
        result *= value
    return result


async def test_context_exposes_live_tools(chat_client_base: SupportsChatGetResponse):
    from agent_framework import FunctionTool

    seen_names: list[str] = []

    @tool(name="inspect_tools", approval_mode="never_require")
    def inspect_tools(ctx: FunctionInvocationContext) -> str:
        assert ctx.tools is not None
        seen_names.extend(t.name for t in ctx.tools if isinstance(t, FunctionTool))
        return "inspected"

    chat_client_base.run_responses = [
        _pte_function_call_response("1", "inspect_tools"),
        _pte_text_response(),
    ]
    await chat_client_base.get_response(
        [Message(role="user", contents=["hi"])],
        options={"tool_choice": "auto", "tools": [inspect_tools]},
    )
    assert "inspect_tools" in seen_names


async def test_add_tools_available_next_iteration(chat_client_base: SupportsChatGetResponse):
    exec_counter = 0

    @tool(name="factorial", approval_mode="never_require")
    def factorial(n: int) -> int:
        nonlocal exec_counter
        exec_counter += 1
        return 120

    @tool(name="load_math", approval_mode="never_require")
    def load_math(ctx: FunctionInvocationContext) -> str:
        ctx.add_tools(factorial)
        return "math tools loaded"

    chat_client_base.function_invocation_configuration["max_iterations"] = 3  # type: ignore[attr-defined]
    chat_client_base.run_responses = [
        _pte_function_call_response("1", "load_math"),
        _pte_function_call_response("2", "factorial", '{"n": 5}'),
        _pte_text_response(),
    ]
    response = await chat_client_base.get_response(
        [Message(role="user", contents=["compute 5!"])],
        options={"tool_choice": "auto", "tools": [load_math]},
    )
    assert exec_counter == 1
    assert response.messages[-1].text == "done"


async def test_add_tools_model_sees_added_tools_in_options(chat_client_base: SupportsChatGetResponse):
    from agent_framework import FunctionTool

    recorded: list[list[str]] = []
    client_cls = type(chat_client_base)
    original = client_cls._get_non_streaming_response

    async def recording(self: Any, *, messages: Any, options: dict[str, Any], **kwargs: Any) -> ChatResponse:
        tools = options.get("tools") or []
        recorded.append([t.name for t in tools if isinstance(t, FunctionTool)])
        return await original(self, messages=messages, options=options, **kwargs)

    @tool(name="load_math", approval_mode="never_require")
    def load_math(ctx: FunctionInvocationContext) -> str:
        ctx.add_tools(_pte_factorial)
        return "loaded"

    chat_client_base.function_invocation_configuration["max_iterations"] = 3  # type: ignore[attr-defined]
    chat_client_base.run_responses = [
        _pte_function_call_response("1", "load_math"),
        _pte_function_call_response("2", "factorial", '{"n": 5}'),
        _pte_text_response(),
    ]
    monkey = pytest.MonkeyPatch()
    monkey.setattr(client_cls, "_get_non_streaming_response", recording)
    try:
        await chat_client_base.get_response(
            [Message(role="user", contents=["compute 5!"])],
            options={"tool_choice": "auto", "tools": [load_math]},
        )
    finally:
        monkey.undo()

    assert recorded[0] == ["load_math"]
    assert "factorial" in recorded[1]


async def test_remove_tools_next_iteration(chat_client_base: SupportsChatGetResponse):
    from agent_framework import FunctionTool

    recorded: list[list[str]] = []
    client_cls = type(chat_client_base)
    original = client_cls._get_non_streaming_response

    async def recording(self: Any, *, messages: Any, options: dict[str, Any], **kwargs: Any) -> ChatResponse:
        tools = options.get("tools") or []
        recorded.append([t.name for t in tools if isinstance(t, FunctionTool)])
        return await original(self, messages=messages, options=options, **kwargs)

    @tool(name="get_weather", approval_mode="never_require")
    def get_weather(location: str) -> str:
        return "sunny"

    @tool(name="drop_weather", approval_mode="never_require")
    def drop_weather(ctx: FunctionInvocationContext) -> str:
        ctx.remove_tools("get_weather")
        return "removed"

    chat_client_base.function_invocation_configuration["max_iterations"] = 3  # type: ignore[attr-defined]
    chat_client_base.run_responses = [
        _pte_function_call_response("1", "drop_weather"),
        _pte_text_response(),
    ]
    monkey = pytest.MonkeyPatch()
    monkey.setattr(client_cls, "_get_non_streaming_response", recording)
    try:
        await chat_client_base.get_response(
            [Message(role="user", contents=["hi"])],
            options={"tool_choice": "auto", "tools": [get_weather, drop_weather]},
        )
    finally:
        monkey.undo()

    assert set(recorded[0]) == {"get_weather", "drop_weather"}
    assert "get_weather" not in recorded[1]


async def test_add_tools_does_not_mutate_caller_tools_list(chat_client_base: SupportsChatGetResponse):
    @tool(name="load_math", approval_mode="never_require")
    def load_math(ctx: FunctionInvocationContext) -> str:
        ctx.add_tools(_pte_factorial)
        return "loaded"

    original_tools: list[Any] = [load_math]
    chat_client_base.function_invocation_configuration["max_iterations"] = 3  # type: ignore[attr-defined]
    chat_client_base.run_responses = [
        _pte_function_call_response("1", "load_math"),
        _pte_text_response(),
    ]
    await chat_client_base.get_response(
        [Message(role="user", contents=["hi"])],
        options={"tool_choice": "auto", "tools": original_tools},
    )
    assert original_tools == [load_math]


async def test_add_tools_persists_across_iterations(chat_client_base: SupportsChatGetResponse):
    from agent_framework import FunctionTool

    recorded: list[list[str]] = []
    client_cls = type(chat_client_base)
    original = client_cls._get_non_streaming_response

    async def recording(self: Any, *, messages: Any, options: dict[str, Any], **kwargs: Any) -> ChatResponse:
        tools = options.get("tools") or []
        recorded.append([t.name for t in tools if isinstance(t, FunctionTool)])
        return await original(self, messages=messages, options=options, **kwargs)

    @tool(name="load_math", approval_mode="never_require")
    def load_math(ctx: FunctionInvocationContext) -> str:
        ctx.add_tools(_pte_factorial)
        return "loaded"

    chat_client_base.function_invocation_configuration["max_iterations"] = 4  # type: ignore[attr-defined]
    chat_client_base.run_responses = [
        _pte_function_call_response("1", "load_math"),
        _pte_function_call_response("2", "factorial", '{"n": 5}'),
        _pte_function_call_response("3", "factorial", '{"n": 3}'),
        _pte_text_response(),
    ]
    monkey = pytest.MonkeyPatch()
    monkey.setattr(client_cls, "_get_non_streaming_response", recording)
    try:
        await chat_client_base.get_response(
            [Message(role="user", contents=["hi"])],
            options={"tool_choice": "auto", "tools": [load_math]},
        )
    finally:
        monkey.undo()

    assert "factorial" in recorded[1]
    assert "factorial" in recorded[2]


async def test_add_tools_through_function_middleware(chat_client_base: SupportsChatGetResponse):
    exec_counter = 0

    class PassthroughMiddleware(FunctionMiddleware):
        async def process(self, context: FunctionInvocationContext, call_next: Any) -> None:
            await call_next()

    @tool(name="factorial", approval_mode="never_require")
    def factorial(n: int) -> int:
        nonlocal exec_counter
        exec_counter += 1
        return 120

    @tool(name="load_math", approval_mode="never_require")
    def load_math(ctx: FunctionInvocationContext) -> str:
        ctx.add_tools(factorial)
        return "loaded"

    chat_client_base.function_invocation_configuration["max_iterations"] = 3  # type: ignore[attr-defined]
    chat_client_base.run_responses = [
        _pte_function_call_response("1", "load_math"),
        _pte_function_call_response("2", "factorial", '{"n": 5}'),
        _pte_text_response(),
    ]
    await chat_client_base.get_response(
        [Message(role="user", contents=["hi"])],
        options={"tool_choice": "auto", "tools": [load_math]},
        middleware=[PassthroughMiddleware()],
    )
    assert exec_counter == 1


async def test_add_tools_with_approval_required_tool(chat_client_base: SupportsChatGetResponse):
    @tool(name="secure_tool", approval_mode="always_require")
    def secure_tool(value: str) -> str:
        return f"secure: {value}"

    @tool(name="load_secure", approval_mode="never_require")
    def load_secure(ctx: FunctionInvocationContext) -> str:
        ctx.add_tools(secure_tool)
        return "loaded"

    chat_client_base.function_invocation_configuration["max_iterations"] = 3  # type: ignore[attr-defined]
    chat_client_base.run_responses = [
        _pte_function_call_response("1", "load_secure"),
        _pte_function_call_response("2", "secure_tool", '{"value": "x"}'),
        _pte_text_response(),
    ]
    response = await chat_client_base.get_response(
        [Message(role="user", contents=["hi"])],
        options={"tool_choice": "auto", "tools": [load_secure]},
    )
    assert any(item.type == "function_approval_request" for msg in response.messages for item in msg.contents)


async def test_add_tools_accepts_plain_callable(chat_client_base: SupportsChatGetResponse):
    exec_counter = 0

    def plain_factorial(n: int) -> int:
        """Compute factorial."""
        nonlocal exec_counter
        exec_counter += 1
        return 120

    @tool(name="load_math", approval_mode="never_require")
    def load_math(ctx: FunctionInvocationContext) -> str:
        ctx.add_tools(plain_factorial)
        return "loaded"

    chat_client_base.function_invocation_configuration["max_iterations"] = 3  # type: ignore[attr-defined]
    chat_client_base.run_responses = [
        _pte_function_call_response("1", "load_math"),
        _pte_function_call_response("2", "plain_factorial", '{"n": 5}'),
        _pte_text_response(),
    ]
    await chat_client_base.get_response(
        [Message(role="user", contents=["hi"])],
        options={"tool_choice": "auto", "tools": [load_math]},
    )
    assert exec_counter == 1


async def test_add_tools_streaming(chat_client_base: SupportsChatGetResponse):
    exec_counter = 0

    @tool(name="factorial", approval_mode="never_require")
    def factorial(n: int) -> int:
        nonlocal exec_counter
        exec_counter += 1
        return 120

    @tool(name="load_math", approval_mode="never_require")
    def load_math(ctx: FunctionInvocationContext) -> str:
        ctx.add_tools(factorial)
        return "loaded"

    chat_client_base.function_invocation_configuration["max_iterations"] = 3  # type: ignore[attr-defined]
    chat_client_base.streaming_responses = [
        [
            ChatResponseUpdate(
                contents=[Content.from_function_call(call_id="1", name="load_math", arguments="{}")],
                role="assistant",
            )
        ],
        [
            ChatResponseUpdate(
                contents=[Content.from_function_call(call_id="2", name="factorial", arguments='{"n": 5}')],
                role="assistant",
            )
        ],
        [ChatResponseUpdate(contents=[Content.from_text("done")], role="assistant", finish_reason="stop")],
    ]
    async for _ in chat_client_base.get_response(
        [Message(role="user", contents=["hi"])],
        stream=True,
        options={"tool_choice": "auto", "tools": [load_math]},
    ):
        pass
    assert exec_counter == 1


def test_add_tools_duplicate_same_object_is_noop():
    @tool(name="dup", approval_mode="never_require")
    def dup(x: int) -> int:
        return x

    ctx = FunctionInvocationContext(function=dup, arguments={}, tools=[dup])
    ctx.add_tools(dup)
    assert ctx.tools is not None
    assert len(ctx.tools) == 1


def test_add_tools_duplicate_name_different_object_raises():
    @tool(name="dup", approval_mode="never_require")
    def dup_a(x: int) -> int:
        return x

    @tool(name="dup", approval_mode="never_require")
    def dup_b(x: int) -> int:
        return x

    ctx = FunctionInvocationContext(function=dup_a, arguments={}, tools=[dup_a])
    with pytest.raises(ValueError):
        ctx.add_tools(dup_b)


def test_add_tools_batch_with_duplicate_is_atomic():
    """A duplicate-name clash partway through a batch must leave the live list unchanged."""

    @tool(name="existing", approval_mode="never_require")
    def existing(x: int) -> int:
        return x

    @tool(name="fresh", approval_mode="never_require")
    def fresh(x: int) -> int:
        return x

    @tool(name="existing", approval_mode="never_require")
    def clashing(x: int) -> int:
        return x

    ctx = FunctionInvocationContext(function=existing, arguments={}, tools=[existing])
    with pytest.raises(ValueError):
        ctx.add_tools([fresh, clashing])
    assert ctx.tools is not None
    # The valid "fresh" tool must not have been committed before the clash raised.
    assert ctx.tools == [existing]


def test_remove_tools_by_name_and_object():
    @tool(name="a", approval_mode="never_require")
    def a(x: int) -> int:
        return x

    @tool(name="b", approval_mode="never_require")
    def b(x: int) -> int:
        return x

    ctx = FunctionInvocationContext(function=a, arguments={}, tools=[a, b])
    ctx.remove_tools("a")
    assert ctx.tools is not None
    assert [t.name for t in ctx.tools] == ["b"]
    ctx.remove_tools(b)
    assert ctx.tools == []


def test_remove_tools_unknown_name_is_noop():
    @tool(name="a", approval_mode="never_require")
    def a(x: int) -> int:
        return x

    ctx = FunctionInvocationContext(function=a, arguments={}, tools=[a])
    ctx.remove_tools("nonexistent")
    assert ctx.tools is not None
    assert [t.name for t in ctx.tools] == ["a"]


def test_progressive_tools_helpers_raise_without_live_tools():
    @tool(name="a", approval_mode="never_require")
    def a(x: int) -> int:
        return x

    ctx = FunctionInvocationContext(function=a, arguments={})
    assert ctx.tools is None
    with pytest.raises(RuntimeError):
        ctx.add_tools(a)
    with pytest.raises(RuntimeError):
        ctx.remove_tools("a")


# endregion
