diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index 398b7356e3..558a47c1ca 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -453,6 +453,7 @@ that manually replay messages own the equivalent rule: do not resend an approval | Middleware termination | Normal non-approval loop stops without a second model call. | `test_terminate_loop_single_function_call`, `test_terminate_loop_multiple_function_calls_one_terminates`, `test_terminate_loop_streaming_single_function_call` | | Maximum iterations | No orphan calls; a final no-tool response or deterministic fallback is returned. | `test_max_iterations_limit`, `test_max_iterations_no_orphaned_function_calls`, `test_max_iterations_makes_final_toolchoice_none_call`, `test_max_iterations_blank_final_fallback_synthesizes_message`, streaming equivalents | | Maximum function calls | Parallel overshoot is bounded after the batch; every executed result group counts even without a `function_result`; blank final responses get fallback content. | `test_max_function_calls_limits_parallel_invocations`, `test_max_function_calls_single_calls_per_iteration`, `test_user_input_request_multiple_contents_propagate`, `test_approval_resume_user_input_counts_toward_function_call_budget`, `test_max_function_calls_blank_final_fallback_synthesizes_message`, streaming equivalent | +| Provider tool content after an active limit | Locally actionable calls and local approval requests returned despite `tool_choice="none"` are removed in both response modes. Provider-executed informational call/result pairs, hosted approval requests, and metadata-only streaming updates remain visible; fallback text never replaces retained transcript content. | `test_function_invocation_limit_drops_unexecutable_tool_content`, `test_streaming_function_invocation_limit_drops_unexecutable_tool_content`, `test_streaming_function_invocation_limit_preserves_metadata_after_tool_content_is_dropped`, `test_function_invocation_limit_preserves_provider_executed_tool_pair`, `test_streaming_function_invocation_limit_preserves_provider_executed_tool_pair`, `test_function_invocation_limit_appends_fallback_after_provider_executed_tool_pair`, `test_streaming_function_invocation_limit_appends_fallback_after_provider_executed_tool_pair`, `test_function_invocation_limit_preserves_hosted_approval_request`, `test_streaming_function_invocation_limit_preserves_hosted_approval_request` | | Conversation continuation | Conversation id updates between iterations and is cleared on stop where required. | `test_conversation_id_updated_in_options_between_tool_iterations`, `test_function_invocation_stop_clears_conversation_id_non_stream`, `test_streaming_function_invocation_stop_clears_conversation_id` | ### History and provider serialization diff --git a/python/packages/core/AGENTS.md b/python/packages/core/AGENTS.md index f14ac146ae..f0ed1f6613 100644 --- a/python/packages/core/AGENTS.md +++ b/python/packages/core/AGENTS.md @@ -157,6 +157,9 @@ agent_framework/ in assistant-role messages, including mixed sibling batches. - Function-call budget accounting counts one unit per executed result group, not per emitted `function_result`, so executions that pause for user input still consume `max_function_calls`. +- Once an invocation limit disables local tools, locally actionable calls and local approval requests are removed + from streaming and final output. Provider-executed informational call/result pairs, hosted approval requests, and + metadata-only stream updates remain visible. - `function_approval_request` and `function_approval_response` are control-plane contents. History providers may retain them in their backing store for audit. The base `HistoryProvider.before_run` filters resolved wrappers from later model replay, but preserves unresolved requests/responses until a terminal result or follow-up request closes diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index 08471275c0..cd342f2fa1 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -1942,6 +1942,25 @@ def _clear_internal_conversation_id(response: ChatResponse[Any]) -> ChatResponse return response +def _is_hosted_tool_approval(content: Any) -> bool: + """Check if a function_approval_request/response is for a hosted tool (e.g. MCP). + + Hosted tool approvals have a server_label in function_call.additional_properties + and should be passed through to the API untouched rather than processed locally. + """ + fc = getattr(content, "function_call", None) + if fc is None: + return False + ap = getattr(fc, "additional_properties", None) + return bool(ap and ap.get("server_label")) + + +def _is_unexecutable_local_tool_content(content: Content) -> bool: + if _is_actionable_function_call(content): + return True + return content.type == "function_approval_request" and not _is_hosted_tool_approval(content) + + def _response_has_visible_content(response: ChatResponse[Any]) -> bool: for message in response.messages: for content in message.contents: @@ -1953,19 +1972,36 @@ def _response_has_visible_content(response: ChatResponse[Any]) -> bool: return False -def _ensure_function_invocation_limit_fallback_response(response: ChatResponse[Any]) -> ChatResponse[Any]: - if _response_has_visible_content(response): - return response +def _response_has_hosted_tool_approval(response: ChatResponse[Any]) -> bool: + return any( + content.type == "function_approval_request" and _is_hosted_tool_approval(content) + for message in response.messages + for content in message.contents + ) + + +def _drop_unexecutable_tool_contents_from_response(response: ChatResponse[Any]) -> None: + for message in response.messages: + if any(_is_unexecutable_local_tool_content(content) for content in message.contents): + message.contents = [ + content for content in message.contents if not _is_unexecutable_local_tool_content(content) + ] + + +def _ensure_function_invocation_limit_fallback_response(response: ChatResponse[Any]) -> bool: + _drop_unexecutable_tool_contents_from_response(response) + if _response_has_visible_content(response) or _response_has_hosted_tool_approval(response): + return False from ._types import Content, Message fallback_content = Content.from_text(_FUNCTION_INVOCATION_LIMIT_FALLBACK_TEXT) - if response.messages: + if response.messages and not response.messages[-1].contents: response.messages[-1].role = "assistant" response.messages[-1].contents = [fallback_content] else: response.messages.append(Message(role="assistant", contents=[fallback_content])) - return response + return True def _function_invocation_limit_fallback_update() -> ChatResponseUpdate: @@ -1978,6 +2014,28 @@ def _function_invocation_limit_fallback_update() -> ChatResponseUpdate: ) +def _update_has_meaningful_metadata(update: ChatResponseUpdate) -> bool: + return any(( + update.author_name is not None, + update.response_id is not None, + update.message_id is not None, + update.conversation_id is not None, + update.model is not None, + update.created_at is not None, + update.finish_reason is not None, + update.continuation_token is not None, + bool(update.additional_properties), + update.raw_representation is not None, + )) + + +def _drop_unexecutable_tool_contents_from_update(update: ChatResponseUpdate) -> ChatResponseUpdate | None: + if not any(_is_unexecutable_local_tool_content(content) for content in update.contents): + return update + update.contents = [content for content in update.contents if not _is_unexecutable_local_tool_content(content)] + return update if update.contents or _update_has_meaningful_metadata(update) else None + + def _extract_tools( options: dict[str, Any] | None, ) -> ToolTypes | Callable[..., Any] | Sequence[ToolTypes | Callable[..., Any]] | None: @@ -1992,19 +2050,6 @@ def _extract_tools( return options.get("tools") if options else None -def _is_hosted_tool_approval(content: Any) -> bool: - """Check if a function_approval_request/response is for a hosted tool (e.g. MCP). - - Hosted tool approvals have a server_label in function_call.additional_properties - and should be passed through to the API untouched rather than processed locally. - """ - fc = getattr(content, "function_call", None) - if fc is None: - return False - ap = getattr(fc, "additional_properties", None) - return bool(ap and ap.get("server_label")) - - def _get_tool_approval_state(invocation_session: AgentSession | None) -> dict[str, Any] | None: """Return the shared tool-approval state bag for the invocation session.""" if invocation_session is None: @@ -2886,7 +2931,7 @@ async def _get_response_with_function_invocation( if options.get("tool_choice") == "none" and _function_call_limit_reached( total_function_calls, max_function_calls ): - response = _ensure_function_invocation_limit_fallback_response(response) + _ensure_function_invocation_limit_fallback_response(response) aggregated_usage = add_usage_details(aggregated_usage, response.usage_details) _update_continuation_state( request_kwargs, @@ -2937,7 +2982,7 @@ async def _get_response_with_function_invocation( client_kwargs=request_kwargs, ), ) - response = _ensure_function_invocation_limit_fallback_response(response) + _ensure_function_invocation_limit_fallback_response(response) aggregated_usage = add_usage_details(aggregated_usage, response.usage_details) _update_continuation_state( request_kwargs, @@ -3011,16 +3056,24 @@ async def _stream_response_with_function_invocation( ), ) await inner_stream + drop_unexecutable_calls = options.get("tool_choice") == "none" and _function_call_limit_reached( + total_function_calls, + max_function_calls, + ) async for update in inner_stream: + if drop_unexecutable_calls: + update = _drop_unexecutable_tool_contents_from_update(update) + if update is None: + continue yield update response = await inner_stream.get_final_response() - response_had_visible_content = _response_has_visible_content(response) function_call_limit_reached = options.get("tool_choice") == "none" and _function_call_limit_reached( total_function_calls, max_function_calls ) + fallback_added = False if function_call_limit_reached: - response = _ensure_function_invocation_limit_fallback_response(response) + fallback_added = _ensure_function_invocation_limit_fallback_response(response) _update_continuation_state( request_kwargs, response, @@ -3033,7 +3086,7 @@ async def _stream_response_with_function_invocation( for message in response.messages for item in message.contents ): - if function_call_limit_reached and not response_had_visible_content: + if fallback_added: yield _function_invocation_limit_fallback_update() return @@ -3082,17 +3135,19 @@ async def _stream_response_with_function_invocation( ) await final_inner_stream async for update in final_inner_stream: + update = _drop_unexecutable_tool_contents_from_update(update) + if update is None: + continue yield update final_response = await final_inner_stream.get_final_response() - final_response_had_visible_content = _response_has_visible_content(final_response) - final_response = _ensure_function_invocation_limit_fallback_response(final_response) + fallback_added = _ensure_function_invocation_limit_fallback_response(final_response) _update_continuation_state( request_kwargs, final_response, session=invocation_session, options=options, ) - if not final_response_had_visible_content: + if fallback_added: yield _function_invocation_limit_fallback_update() @overload diff --git a/python/packages/core/tests/core/test_function_invocation_logic.py b/python/packages/core/tests/core/test_function_invocation_logic.py index 07b7d6ae7d..d0db1c7c4e 100644 --- a/python/packages/core/tests/core/test_function_invocation_logic.py +++ b/python/packages/core/tests/core/test_function_invocation_logic.py @@ -2,7 +2,7 @@ import asyncio from collections.abc import AsyncIterable, Awaitable, Callable, Sequence -from typing import Any +from typing import Any, Literal import pytest @@ -96,6 +96,88 @@ def _finalize(updates: Sequence[ChatResponseUpdate]) -> ChatResponse: chat_client_base._get_streaming_response = _get_streaming_response +def _post_limit_local_tool_content( + content_type: Literal["function_call", "function_approval_request"], +) -> Content: + function_call = Content.from_function_call(call_id="local_call_2", name="lookup", arguments='{"key": "b"}') + if content_type == "function_call": + return function_call + return Content.from_function_approval_request(id="local_approval_2", function_call=function_call) + + +def _post_limit_informational_transcript() -> list[Content]: + return [ + Content.from_function_call( + call_id="provider_call_2", + name="provider_lookup", + arguments='{"key": "b"}', + informational_only=True, + ), + Content.from_function_result(call_id="provider_call_2", result="Value for b"), + Content.from_text("Provider completed the lookup."), + ] + + +def _post_limit_hosted_approval_request() -> Content: + function_call = Content.from_function_call( + call_id="hosted_call_2", + name="hosted_lookup", + arguments='{"key": "b"}', + additional_properties={"server_label": "hosted_server"}, + ) + return Content.from_function_approval_request(id="hosted_approval_2", function_call=function_call) + + +def _configure_function_invocation_limit( + chat_client_base: Any, + limit_type: Literal["max_function_calls", "max_iterations"], +) -> None: + if limit_type == "max_function_calls": + chat_client_base.function_invocation_configuration["max_iterations"] = 10 + chat_client_base.function_invocation_configuration["max_function_calls"] = 1 + else: + chat_client_base.function_invocation_configuration["max_iterations"] = 1 + + +def _force_tool_content_tool_choice_none_stream( + chat_client_base: Any, + *, + contents: Sequence[Content], + additional_properties: dict[str, Any] | None = None, + finish_reason: Literal["stop", "length", "tool_calls", "content_filter"] | None = None, +) -> None: + original_get_streaming_response = chat_client_base._get_streaming_response + + def _get_streaming_response( + *, + messages: Sequence[Message], + options: dict[str, Any], + **kwargs: Any, + ) -> ResponseStream[ChatResponseUpdate, ChatResponse]: + if options.get("tool_choice") != "none": + return original_get_streaming_response(messages=messages, options=options, **kwargs) + + updates = ( + ChatResponseUpdate( + contents=list(contents), + role="assistant", + additional_properties=additional_properties, + finish_reason=finish_reason, + ), + ) + + async def _stream() -> AsyncIterable[ChatResponseUpdate]: + for update in updates: + yield update + + def _finalize(stream_updates: Sequence[ChatResponseUpdate]) -> ChatResponse: + return ChatResponse.from_updates(stream_updates, output_format_type=options.get("response_format")) + + return ResponseStream(_stream(), finalizer=_finalize) + + chat_client_base._get_streaming_response = _get_streaming_response + + async def test_base_client_with_function_calling(chat_client_base: SupportsChatGetResponse): exec_counter = 0 @@ -3672,6 +3754,413 @@ def lookup_func(key: str) -> str: assert updates[-1].text == _EXPECTED_FUNCTION_INVOCATION_LIMIT_FALLBACK_TEXT +@pytest.mark.parametrize("limit_type", ["max_function_calls", "max_iterations"]) +@pytest.mark.parametrize("content_type", ["function_call", "function_approval_request"]) +async def test_function_invocation_limit_drops_unexecutable_tool_content( + chat_client_base: SupportsChatGetResponse, + limit_type: Literal["max_function_calls", "max_iterations"], + content_type: Literal["function_call", "function_approval_request"], +) -> None: + """Tool content returned after the active limit is not exposed without a terminal result.""" + _force_blank_tool_choice_none_fallback( + chat_client_base, + final_contents=[_post_limit_local_tool_content(content_type)], + ) + 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 = [ # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + ChatResponse( + messages=Message( + role="assistant", + contents=[Content.from_function_call(call_id="call_1", name="lookup", arguments='{"key": "a"}')], + ) + ) + ] + _configure_function_invocation_limit(chat_client_base, limit_type) + + response = await chat_client_base.get_response( + [Message(role="user", contents=["look up key"])], + options={"tool_choice": "auto", "tools": [lookup_func]}, + ) + + function_call_ids = { + content.call_id + for message in response.messages + for content in message.contents + if content.type == "function_call" + } + approval_request_ids = { + content.id + for message in response.messages + for content in message.contents + if content.type == "function_approval_request" + } + assert exec_counter == 1 + assert function_call_ids == {"call_1"} + assert not approval_request_ids + assert response.messages[-1].text == _EXPECTED_FUNCTION_INVOCATION_LIMIT_FALLBACK_TEXT + + +@pytest.mark.parametrize("limit_type", ["max_function_calls", "max_iterations"]) +@pytest.mark.parametrize("content_type", ["function_call", "function_approval_request"]) +async def test_streaming_function_invocation_limit_drops_unexecutable_tool_content( + chat_client_base: SupportsChatGetResponse, + limit_type: Literal["max_function_calls", "max_iterations"], + content_type: Literal["function_call", "function_approval_request"], +) -> None: + """Tool content returned after the active limit is not exposed without a terminal result.""" + _force_tool_content_tool_choice_none_stream( + chat_client_base, + contents=[_post_limit_local_tool_content(content_type)], + ) + 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.streaming_responses = [ # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + [ + ChatResponseUpdate( + contents=[Content.from_function_call(call_id="call_1", name="lookup", arguments='{"key": "a"}')], + role="assistant", + ), + ], + ] + _configure_function_invocation_limit(chat_client_base, limit_type) + + updates = [ + update + async for update in chat_client_base.get_response( + [Message(role="user", contents=["look up key"])], + options={"tool_choice": "auto", "tools": [lookup_func]}, + stream=True, + ) + ] + + function_call_ids = { + content.call_id for update in updates for content in update.contents if content.type == "function_call" + } + function_result_ids = { + content.call_id for update in updates for content in update.contents if content.type == "function_result" + } + approval_request_ids = { + content.id for update in updates for content in update.contents if content.type == "function_approval_request" + } + assert exec_counter == 1 + assert function_call_ids == {"call_1"} + assert function_result_ids == {"call_1"} + assert not approval_request_ids + assert updates[-1].text == _EXPECTED_FUNCTION_INVOCATION_LIMIT_FALLBACK_TEXT + + +@pytest.mark.parametrize("limit_type", ["max_function_calls", "max_iterations"]) +@pytest.mark.parametrize("content_type", ["function_call", "function_approval_request"]) +async def test_streaming_function_invocation_limit_preserves_metadata_after_tool_content_is_dropped( + chat_client_base: SupportsChatGetResponse, + limit_type: Literal["max_function_calls", "max_iterations"], + content_type: Literal["function_call", "function_approval_request"], +) -> None: + """Metadata-only chunks survive when their unexecutable function call content is removed.""" + _force_tool_content_tool_choice_none_stream( + chat_client_base, + contents=[_post_limit_local_tool_content(content_type)], + additional_properties={"provider_metadata": "keep"}, + finish_reason="stop", + ) + + @tool(name="lookup", approval_mode="never_require") + def lookup_func(key: str) -> str: + return f"Value for {key}" + + chat_client_base.streaming_responses = [ # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + [ + ChatResponseUpdate( + contents=[Content.from_function_call(call_id="call_1", name="lookup", arguments='{"key": "a"}')], + role="assistant", + ), + ], + ] + _configure_function_invocation_limit(chat_client_base, limit_type) + + updates = [ + update + async for update in chat_client_base.get_response( + [Message(role="user", contents=["look up key"])], + options={"tool_choice": "auto", "tools": [lookup_func]}, + stream=True, + ) + ] + metadata_updates = [ + update + for update in updates + if update.additional_properties == {"provider_metadata": "keep"} and update.finish_reason == "stop" + ] + assert len(metadata_updates) == 1 + assert metadata_updates[0].contents == [] + + +@pytest.mark.parametrize("limit_type", ["max_function_calls", "max_iterations"]) +async def test_function_invocation_limit_preserves_provider_executed_tool_pair( + chat_client_base: SupportsChatGetResponse, + limit_type: Literal["max_function_calls", "max_iterations"], +) -> None: + """Provider-executed tool transcripts remain complete after the local invocation limit.""" + provider_transcript = _post_limit_informational_transcript() + final_contents = [_post_limit_local_tool_content("function_call"), *provider_transcript] + _force_blank_tool_choice_none_fallback(chat_client_base, final_contents=final_contents) + 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 = [ # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + ChatResponse( + messages=Message( + role="assistant", + contents=[Content.from_function_call(call_id="call_1", name="lookup", arguments='{"key": "a"}')], + ) + ) + ] + _configure_function_invocation_limit(chat_client_base, limit_type) + + response = await chat_client_base.get_response( + [Message(role="user", contents=["look up key"])], + options={"tool_choice": "auto", "tools": [lookup_func]}, + ) + + assert exec_counter == 1 + assert response.messages[-1].contents == provider_transcript + assert response.messages[-1].text == "Provider completed the lookup." + + +@pytest.mark.parametrize("limit_type", ["max_function_calls", "max_iterations"]) +async def test_streaming_function_invocation_limit_preserves_provider_executed_tool_pair( + chat_client_base: SupportsChatGetResponse, + limit_type: Literal["max_function_calls", "max_iterations"], +) -> None: + """Streaming preserves provider-executed calls with their paired results after the local limit.""" + provider_transcript = _post_limit_informational_transcript() + final_contents = [_post_limit_local_tool_content("function_call"), *provider_transcript] + _force_tool_content_tool_choice_none_stream(chat_client_base, contents=final_contents) + 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.streaming_responses = [ # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + [ + ChatResponseUpdate( + contents=[Content.from_function_call(call_id="call_1", name="lookup", arguments='{"key": "a"}')], + role="assistant", + ), + ], + ] + _configure_function_invocation_limit(chat_client_base, limit_type) + + updates = [ + update + async for update in chat_client_base.get_response( + [Message(role="user", contents=["look up key"])], + options={"tool_choice": "auto", "tools": [lookup_func]}, + stream=True, + ) + ] + provider_contents = [ + content for update in updates for content in update.contents if content.call_id == "provider_call_2" + ] + + assert exec_counter == 1 + assert [content.type for content in provider_contents] == ["function_call", "function_result"] + assert not any(content.call_id == "local_call_2" for update in updates for content in update.contents) + assert any(update.text == "Provider completed the lookup." for update in updates) + assert not any(update.text == _EXPECTED_FUNCTION_INVOCATION_LIMIT_FALLBACK_TEXT for update in updates) + + +async def test_function_invocation_limit_appends_fallback_after_provider_executed_tool_pair( + chat_client_base: SupportsChatGetResponse, +) -> None: + """Fallback text is added without replacing a provider-owned call/result pair.""" + provider_transcript = _post_limit_informational_transcript()[:2] + final_contents = [_post_limit_local_tool_content("function_call"), *provider_transcript] + _force_blank_tool_choice_none_fallback(chat_client_base, final_contents=final_contents) + + @tool(name="lookup", approval_mode="never_require") + def lookup_func(key: str) -> str: + return f"Value for {key}" + + chat_client_base.run_responses = [ # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + ChatResponse( + messages=Message( + role="assistant", + contents=[Content.from_function_call(call_id="call_1", name="lookup", arguments='{"key": "a"}')], + ) + ) + ] + _configure_function_invocation_limit(chat_client_base, "max_function_calls") + + response = await chat_client_base.get_response( + [Message(role="user", contents=["look up key"])], + options={"tool_choice": "auto", "tools": [lookup_func]}, + ) + + assert response.messages[-2].contents == provider_transcript + assert response.messages[-1].text == _EXPECTED_FUNCTION_INVOCATION_LIMIT_FALLBACK_TEXT + + +async def test_streaming_function_invocation_limit_appends_fallback_after_provider_executed_tool_pair( + chat_client_base: SupportsChatGetResponse, +) -> None: + """Streaming emits fallback text after preserving a provider-owned call/result pair.""" + provider_transcript = _post_limit_informational_transcript()[:2] + final_contents = [_post_limit_local_tool_content("function_call"), *provider_transcript] + _force_tool_content_tool_choice_none_stream(chat_client_base, contents=final_contents) + + @tool(name="lookup", approval_mode="never_require") + def lookup_func(key: str) -> str: + return f"Value for {key}" + + chat_client_base.streaming_responses = [ # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + [ + ChatResponseUpdate( + contents=[Content.from_function_call(call_id="call_1", name="lookup", arguments='{"key": "a"}')], + role="assistant", + ), + ], + ] + _configure_function_invocation_limit(chat_client_base, "max_function_calls") + + updates = [ + update + async for update in chat_client_base.get_response( + [Message(role="user", contents=["look up key"])], + options={"tool_choice": "auto", "tools": [lookup_func]}, + stream=True, + ) + ] + provider_contents = [ + content for update in updates for content in update.contents if content.call_id == "provider_call_2" + ] + + assert [content.type for content in provider_contents] == ["function_call", "function_result"] + assert not any(content.call_id == "local_call_2" for update in updates for content in update.contents) + assert updates[-1].text == _EXPECTED_FUNCTION_INVOCATION_LIMIT_FALLBACK_TEXT + + +@pytest.mark.parametrize("limit_type", ["max_function_calls", "max_iterations"]) +async def test_function_invocation_limit_preserves_hosted_approval_request( + chat_client_base: SupportsChatGetResponse, + limit_type: Literal["max_function_calls", "max_iterations"], +) -> None: + """Hosted approvals remain available to the caller after the local invocation limit.""" + hosted_approval = _post_limit_hosted_approval_request() + final_contents = [_post_limit_local_tool_content("function_approval_request"), hosted_approval] + _force_blank_tool_choice_none_fallback(chat_client_base, final_contents=final_contents) + 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 = [ # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + ChatResponse( + messages=Message( + role="assistant", + contents=[Content.from_function_call(call_id="call_1", name="lookup", arguments='{"key": "a"}')], + ) + ) + ] + _configure_function_invocation_limit(chat_client_base, limit_type) + + response = await chat_client_base.get_response( + [Message(role="user", contents=["look up key"])], + options={"tool_choice": "auto", "tools": [lookup_func]}, + ) + + assert exec_counter == 1 + assert any( + content.type == "function_approval_request" and content.id == "hosted_approval_2" + for message in response.messages + for content in message.contents + ) + assert not any( + content.type == "function_approval_request" and content.id == "local_approval_2" + for message in response.messages + for content in message.contents + ) + assert not any( + content.text == _EXPECTED_FUNCTION_INVOCATION_LIMIT_FALLBACK_TEXT + for message in response.messages + for content in message.contents + ) + + +@pytest.mark.parametrize("limit_type", ["max_function_calls", "max_iterations"]) +async def test_streaming_function_invocation_limit_preserves_hosted_approval_request( + chat_client_base: SupportsChatGetResponse, + limit_type: Literal["max_function_calls", "max_iterations"], +) -> None: + """Streaming keeps hosted approvals actionable after the local invocation limit.""" + hosted_approval = _post_limit_hosted_approval_request() + final_contents = [_post_limit_local_tool_content("function_approval_request"), hosted_approval] + _force_tool_content_tool_choice_none_stream(chat_client_base, contents=final_contents) + 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.streaming_responses = [ # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + [ + ChatResponseUpdate( + contents=[Content.from_function_call(call_id="call_1", name="lookup", arguments='{"key": "a"}')], + role="assistant", + ), + ], + ] + _configure_function_invocation_limit(chat_client_base, limit_type) + + updates = [ + update + async for update in chat_client_base.get_response( + [Message(role="user", contents=["look up key"])], + options={"tool_choice": "auto", "tools": [lookup_func]}, + stream=True, + ) + ] + + assert exec_counter == 1 + assert any( + content.type == "function_approval_request" and content.id == "hosted_approval_2" + for update in updates + for content in update.contents + ) + assert not any( + content.type == "function_approval_request" and content.id == "local_approval_2" + for update in updates + for content in update.contents + ) + assert not any(update.text == _EXPECTED_FUNCTION_INVOCATION_LIMIT_FALLBACK_TEXT for update in updates) + + 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