diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index a292383ac7a..35f28d05bc0 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -491,10 +491,11 @@ def __init__( max_invocations: The maximum number of times this function can be invoked across the **lifetime of this tool instance**. If None (default), there is no limit. Should be at least 1. If the tool is called multiple - times in one iteration, those will execute, after that it will stop working. For example, - if max_invocations is 3 and the tool is called 5 times in a single iteration, - these will complete, but any subsequent calls to the tool (in the same or future iterations) - will raise a ToolException. + times in one iteration, those will execute, after that it will stop + working. For example, if max_invocations is 3 and the tool is called 5 + times in a single iteration, these will complete, but any subsequent + calls to the tool (in the same or future iterations) will raise a + ToolException. .. note:: This counter lives on the tool instance and is never automatically @@ -505,8 +506,8 @@ def __init__( ``FunctionInvocationConfiguration["max_function_calls"]`` for per-request limits instead. - max_invocation_exceptions: The maximum number of exceptions allowed during invocations. - If None, there is no limit. Should be at least 1. + max_invocation_exceptions: The maximum number of exceptions allowed + during invocations. If None, there is no limit. Should be at least 1. additional_properties: Additional properties to set on the function. func: The function to wrap. When ``None``, creates a declaration-only tool that has no implementation. Declaration-only tools are useful when you want @@ -540,6 +541,7 @@ def __init__( :meth:`invoke` returns the wrapped function's raw return value. Depending on your function, it may be easiest to just do the serialization directly in the function body rather than providing a custom ``result_parser``. + **kwargs: Additional keyword arguments. """ # Core attributes (formerly from BaseTool) @@ -1233,7 +1235,7 @@ def to_dict(self, *, exclude: set[str] | None = None, exclude_none: bool = True) as_dict = super().to_dict(exclude=exclude, exclude_none=exclude_none) if (exclude and "input_model" in exclude) or not self.input_model: return as_dict - as_dict["input_model"] = self.parameters() # Use cached parameters() + as_dict["input_model"] = self.parameters() return as_dict @@ -1700,6 +1702,14 @@ class FunctionInvocationConfiguration(TypedDict, total=False): - ``include_detailed_errors``: Whether to include exception details in the function result returned to the model. Exception text may contain sensitive information regardless of its source, so enable this only for a trusted channel. + - ``allow_concurrent_invocation``: Dictates whether multiple tool calls in a + single message batch are executed concurrently (``True``, default) or + one-by-one (``False``). When set to ``False``, tools run sequentially. If a + call requests termination (e.g., via middleware), the loop immediately stops + dequeuing subsequent calls and safely skips them, ensuring provider + continuation history remains resolved. Ordinary tool failures do not stop + the loop; they are converted to error results and the next tool in the + batch is executed. Note: ``max_iterations``, ``max_function_calls``, and ``max_duration_seconds`` @@ -1732,6 +1742,7 @@ class FunctionInvocationConfiguration(TypedDict, total=False): terminate_on_unknown_calls: bool additional_tools: Sequence[FunctionTool] include_detailed_errors: bool + allow_concurrent_invocation: bool def normalize_function_invocation_configuration( @@ -1746,6 +1757,7 @@ def normalize_function_invocation_configuration( "terminate_on_unknown_calls": False, "additional_tools": [], "include_detailed_errors": False, + "allow_concurrent_invocation": True, } if config: normalized.update(config) @@ -1794,6 +1806,11 @@ def _function_execution_error_result( ) +def _is_server_managed_tool(tool: FunctionTool) -> bool: + """Check if a tool is server-managed and should not be executed locally.""" + return bool(tool.additional_properties and tool.additional_properties.get("server_label")) + + def _function_argument_validation_error_result( function_call: Content, exception: _FunctionArgumentValidationError, @@ -2156,7 +2173,7 @@ def _get_tool_map( return { tool_item.name: tool_item for tool_item in _ensure_unique_tool_names(tools) - if isinstance(tool_item, FunctionTool) + if isinstance(tool_item, FunctionTool) and not _is_server_managed_tool(tool_item) } @@ -2200,6 +2217,20 @@ async def _execute_single_function_call( if (source_tool is not None and source_tool.declaration_only) or source_function_call.name in additional_tool_names: return [_as_user_input_pause(source_function_call)], False + tool_name = source_function_call.name + if ( + source_function_call.additional_properties and source_function_call.additional_properties.get("server_label") + ) or tool_name not in tool_map: + exc = KeyError(f'Function "{tool_name}" not found.') + return [ + Content.from_function_result( + call_id=source_function_call.call_id, # type: ignore[arg-type] + result=f'Error: Requested function "{tool_name}" not found.', + exception=str(exc), + additional_properties=source_function_call.additional_properties, + ) + ], False + try: # A run-persistence gate defers only the gated run's own persistence; nested # agent runs persist inline at their own boundaries. Run-identity ownership @@ -2218,11 +2249,10 @@ async def _execute_single_function_call( live_tools=live_tools, host_payload_budget=host_payload_budget, ) - return [result], False + return [result], False except MiddlewareTermination as exc: if isinstance(exc.result, Content): return [exc.result], True - source_function_call = _underlying_function_call(function_call) return [ Content.from_function_result( call_id=source_function_call.call_id, # type: ignore[arg-type] @@ -2230,7 +2260,6 @@ async def _execute_single_function_call( ) ], True except UserInputRequiredException as exc: - source_function_call = _underlying_function_call(function_call) call_id = source_function_call.call_id propagated_contents = [item for item in exc.contents if isinstance(item, Content)] if exc.contents else [] for item in propagated_contents: @@ -2314,7 +2343,8 @@ async def _try_execute_function_call_groups( unknown_call_found = False unknown_call_name: str | None = None for function_call in actionable_calls: - function_name = function_call.name + function_name = _underlying_function_call(function_call).name + logger.debug( "Checking function call: type=%s, name=%s, in approval_tools=%s", function_call.type, @@ -2338,9 +2368,10 @@ async def _try_execute_function_call_groups( # executable siblings remain hidden until the approval batch resumes. logger.debug("Returning visible function_approval_request contents and storing already-approved requests") visible_requests: list[Content] = [] - already_approved_requests: list[Content] = [] + already_approved_requests: list[tuple[int, Content]] = [] + batch_id = str(uuid4()) pause_groups: list[list[Content]] = [] - for function_call in function_calls: + for idx, function_call in enumerate(function_calls): if function_call.type != "function_call": continue tool_name = function_call.name @@ -2355,6 +2386,9 @@ async def _try_execute_function_call_groups( id=function_call.id or function_call.call_id, # type: ignore[arg-type] function_call=function_call, ) + approval_request.additional_properties["original_index"] = idx + approval_request.additional_properties["batch_id"] = batch_id + tool_name = function_call.name if tool_name is None: visible_requests.append(approval_request) pause_groups.append([approval_request]) @@ -2368,7 +2402,7 @@ async def _try_execute_function_call_groups( visible_requests.append(approval_request) pause_groups.append([approval_request]) continue - already_approved_requests.append(approval_request) + already_approved_requests.append((idx, approval_request)) _store_already_approved_approval_requests( approval_session, visible_requests, @@ -2390,11 +2424,16 @@ async def _try_execute_function_call_groups( # Only a fully executable batch reaches this point; run calls concurrently but retain per-call result groups. # Create each task inside a copied context so the active agent span is # preserved for every parallel tool invocation. - execution_tasks = [ - contextvars.copy_context().run( + + allow_concurrent = config.get("allow_concurrent_invocation", True) + execution_results: list[tuple[list[Content], bool]] = [] + + def _create_execution_task(call: Content) -> asyncio.Task[tuple[list[Content], bool]]: + ctx = contextvars.copy_context() + return ctx.run( asyncio.create_task, _execute_single_function_call( - function_call, + call, custom_args=custom_args, config=config, tool_map=tool_map, @@ -2404,22 +2443,35 @@ async def _try_execute_function_call_groups( host_payload_budget=host_payload_budget, ), ) - for function_call in function_calls - ] - try: - execution_results = await asyncio.gather(*execution_tasks) - except BaseException: - # A loud escape from one call (e.g. MiddlewareFailure aborting the run - # fail-closed) fails the whole batch: cancel in-flight siblings and wait for - # them so no new tool work starts after the loop is abandoned. Cancellation - # is cooperative — a synchronous tool body already running in a worker thread - # (asyncio.to_thread) cannot be interrupted and may complete its side effects, - # but its result is discarded with the batch and never reaches the transcript, - # the model, or history. - for task in execution_tasks: - task.cancel() - await asyncio.gather(*execution_tasks, return_exceptions=True) - raise + + if allow_concurrent: + tasks = [_create_execution_task(call) for call in function_calls] + try: + execution_results = await asyncio.gather(*tasks) + except BaseException: + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + raise + else: + for idx, call in enumerate(function_calls): + task = _create_execution_task(call) + try: + res = await task + except BaseException: + task.cancel() + raise + + execution_results.append(res) + if res[1]: + for skipped in function_calls[idx + 1 :]: + skipped_call = _underlying_function_call(skipped) + skipped_result = Content.from_function_result( + call_id=skipped_call.call_id, # type: ignore[arg-type] + result="Skipped: a prior tool call in this batch requested termination.", + ) + execution_results.append(([skipped_result], False)) + break should_terminate = any(terminate for _, terminate in execution_results) return [result_contents for result_contents, _ in execution_results], should_terminate @@ -2469,15 +2521,22 @@ async def _execute_function_calls( custom_args: dict[str, Any], function_calls: list[Content], options: dict[str, Any] | None, - config: FunctionInvocationConfiguration, + config_provider: Callable[[], FunctionInvocationConfiguration], invocation_session: AgentSession | None = None, approval_session_is_authoritative: bool = True, middleware_pipeline: FunctionMiddlewarePipeline | None = None, host_payload_budget: _FunctionResultPayloadBudget | None = None, ) -> _FunctionExecutionBatch: + config = config_provider() + run_config = cast( + "FunctionInvocationConfiguration", + dict(config) if config else {}, + ) + tools = _extract_tools(options) if not tools: return _FunctionExecutionBatch(result_groups=[]) + result_groups, should_terminate = await _try_execute_function_call_groups( custom_args=custom_args, function_calls=function_calls, @@ -2485,7 +2544,7 @@ async def _execute_function_calls( invocation_session=invocation_session, approval_session_is_authoritative=approval_session_is_authoritative, middleware_pipeline=middleware_pipeline, - config=config, + config=run_config, host_payload_budget=host_payload_budget, ) return _FunctionExecutionBatch( @@ -2828,6 +2887,12 @@ def _bind_approval_response_to_pending_request( additional_properties=rebound_properties, raw_representation=response.raw_representation, ) + if request.additional_properties: + if "original_index" in request.additional_properties: + rebound.additional_properties["original_index"] = request.additional_properties["original_index"] + if "batch_id" in request.additional_properties: + rebound.additional_properties["batch_id"] = request.additional_properties["batch_id"] + if consume: pending.pop(request_key, None) _save_pending_approval_requests(invocation_session, pending) @@ -2871,7 +2936,7 @@ def _bind_approval_responses_to_pending_requests( def _store_already_approved_approval_requests( invocation_session: AgentSession | None, visible_approval_requests: Sequence[Content], - already_approved_requests: Sequence[Content], + already_approved_requests: Sequence[tuple[int, Content] | Content], ) -> None: """Store hidden already-approved requests keyed by the visible approvals that resume the batch.""" if not already_approved_requests: @@ -2879,15 +2944,33 @@ def _store_already_approved_approval_requests( state = _get_tool_approval_state(invocation_session) if state is None: return - visible_ids = [request.id for request in visible_approval_requests if request.id] + visible_ids: list[str] = [request.id for request in visible_approval_requests if request.id] if not visible_ids: return existing_groups = state.get(_ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY) - pending_groups = list(cast(list[Any], existing_groups)) if isinstance(existing_groups, list) else [] + + pending_groups: list[dict[str, Any]] = ( + cast("list[dict[str, Any]]", existing_groups) if isinstance(existing_groups, list) else [] + ) + + serialized_requests: list[dict[str, Any]] = [] + request_indices: list[int] = [] + for i, item in enumerate(already_approved_requests): + idx: int + request: Content + if isinstance(item, tuple): + idx, request = item + else: + idx, request = i, item + + serialized_requests.append(request.to_dict()) + request_indices.append(idx) + pending_groups.append({ "approval_request_ids": visible_ids, - "approval_requests": [request.to_dict() for request in already_approved_requests], + "approval_requests": serialized_requests, + "approval_request_indices": request_indices, }) state[_ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY] = pending_groups @@ -2896,7 +2979,7 @@ def _pop_already_approved_approval_responses( invocation_session: AgentSession | None, approval_response_ids: set[str], ) -> list[Content]: - """Pop already-approved requests for the visible approval ids being answered.""" + """Pop already-approved requests for the visible approval ids being answered, preserving original order.""" if not approval_response_ids: return [] state = _get_tool_approval_state(invocation_session) @@ -2905,27 +2988,55 @@ def _pop_already_approved_approval_responses( raw_groups = state.get(_ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY, []) if not isinstance(raw_groups, list): return [] - typed_groups = cast(list[Any], raw_groups) + typed_groups = cast("list[Mapping[str, Any]]", raw_groups) - responses: list[Content] = [] - remaining_groups: list[Any] = [] + tagged_responses: list[tuple[int, Content]] = [] + remaining_groups: list[Mapping[str, Any]] = [] for raw_group in typed_groups: if not isinstance(raw_group, Mapping): continue - group = cast(Mapping[str, Any], raw_group) + group = raw_group raw_ids = group.get("approval_request_ids") - group_ids: set[str] = {str(item) for item in cast(list[Any], raw_ids)} if isinstance(raw_ids, list) else set() - if group_ids.isdisjoint(approval_response_ids): - remaining_groups.append(raw_group) + group_ids: set[str] = {str(item) for item in cast("list[Any]", raw_ids)} if isinstance(raw_ids, list) else set() + + answered_ids = group_ids.intersection(approval_response_ids) + if not answered_ids: + remaining_groups.append(group) + continue + + remaining_visible_ids = list(group_ids - answered_ids) + + if remaining_visible_ids: + updated_group = dict(group) + updated_group["approval_request_ids"] = remaining_visible_ids + remaining_groups.append(updated_group) continue + raw_requests = group.get("approval_requests") if not isinstance(raw_requests, list): continue - for raw_request in cast(list[Any], raw_requests): - request = _content_from_state(raw_request) + + raw_indices = group.get("approval_request_indices") + indices = cast("list[int]", raw_indices) if isinstance(raw_indices, list) else [] + + for i, raw_request in enumerate(cast(list[Any], raw_requests)): + if isinstance(raw_request, Mapping) and "original_index" in raw_request: + req_map = cast("Mapping[str, Any]", raw_request) + original_index: int = int(req_map["original_index"]) + request_dict: dict[str, Any] = dict(req_map) + request_dict.pop("original_index", None) + request = _content_from_state(request_dict) + else: + original_index = indices[i] if i < len(indices) else 0 + request = _content_from_state(raw_request) + if request is None or request.type != "function_approval_request": continue - responses.append(request.to_function_approval_response(approved=True)) + tagged_responses.append((original_index, request.to_function_approval_response(approved=True))) + + tagged_responses.sort(key=lambda item: item[0]) + responses = [content for _, content in tagged_responses] + if remaining_groups: state[_ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY] = remaining_groups else: @@ -4019,6 +4130,7 @@ async def _resolve_approval_responses( approval_session_is_authoritative: bool = True, middleware_pipeline: FunctionMiddlewarePipeline | None = None, settle_dangling_calls: Callable[[Sequence[Content]], Awaitable[None]] | None = None, + allow_concurrent_invocation: bool = True, ) -> _FunctionProcessingResult: """Resolve inbound approval responses before the next model call. @@ -4085,6 +4197,71 @@ async def _resolve_approval_responses( responses_to_execute = [ response for response in pending_approval_responses.values() if _is_approval_granted(response.approved) ] + + # Sort by original batch index to restore model order (critical for sequential mode) + responses_to_execute.sort( + key=lambda c: c.additional_properties.get("original_index", 0) if c.additional_properties else 0 + ) + + # if there are pending approval with a lower original_index. + # we must hold back the current approval calls until earlier ones are resolved. + if not allow_concurrent_invocation: + pending_unanswered = _load_pending_approval_requests(invocation_session) + if pending_unanswered: + min_pending_index = min( + ( + req.additional_properties.get("original_index", 0) + for req in pending_unanswered.values() + if req.additional_properties and req.additional_properties.get("batch_id") is not None + ), + default=None, + ) + + # if we have pending request filter out any approved response that come after the earliest pending request + + if min_pending_index is not None: + executable_now: list[Content] = [] + held_back: list[Content] = [] + for res in responses_to_execute: + res_index = res.additional_properties.get("original_index", 0) if res.additional_properties else 0 + if res_index > min_pending_index: + held_back.append(res) + else: + executable_now.append(res) + + if held_back: + # Convert held-back approval responses back to approval requests + from ._types import Content + + held_back_request: list[Content] = [] + for res in held_back: + if res.function_call is not None and res.id is not None: + req_id = res.additional_properties.get(_APPROVAL_REQUEST_ID_KEY, res.id) + held_back_request.append( + Content.from_function_approval_request( + id=req_id, + function_call=res.function_call, + additional_properties=res.additional_properties, + ) + ) + + # Store them as already-approved requests keyed by the visible pending approvals + # so they are restored and executed when earlier approvals are answered. + visible_pending = list(pending_unanswered.values()) + _store_already_approved_approval_requests(invocation_session, visible_pending, held_back_request) + + # Remove held-back responses from current execution path and pending dict + for res in held_back: + if res.id in pending_approval_responses: + del pending_approval_responses[res.id] + + responses_to_execute = executable_now + + # If nothing is executable now, return to wait for earlier approvals + if not responses_to_execute: + _remove_unanswered_approval_batches_from_model_input(prepared_messages) + return _FunctionProcessingResult(errors_in_a_row=errors_in_a_row, action="return") + responses_not_granted = [ response for response in pending_approval_responses.values() if not _is_approval_granted(response.approved) ] @@ -4092,6 +4269,7 @@ async def _resolve_approval_responses( middleware_pipeline._notify_approval_responses( # pyright: ignore[reportPrivateUsage] responses_not_granted, session=invocation_session ) + execution_result_groups: list[list[Content]] = [] should_terminate = False reached_error_limit = False @@ -4440,6 +4618,7 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non approval_session_is_authoritative=approval_session_is_authoritative, middleware_pipeline=middleware_pipeline, settle_dangling_calls=settle_approval_replay_calls, + allow_concurrent_invocation=self.function_invocation_configuration.get("allow_concurrent_invocation", True), ) function_call_messages.extend(approval_processing.response_messages) errors_in_a_row = approval_processing.errors_in_a_row @@ -4653,6 +4832,7 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non approval_session_is_authoritative=approval_session_is_authoritative, middleware_pipeline=middleware_pipeline, settle_dangling_calls=settle_approval_replay_calls, + allow_concurrent_invocation=self.function_invocation_configuration.get("allow_concurrent_invocation", True), ) errors_in_a_row = approval_processing.errors_in_a_row total_function_calls = _record_function_calls( @@ -4985,19 +5165,22 @@ def get_response( approval_session_is_authoritative = False # Bind one executor with the run's custom arguments, middleware, configuration, and session. + mutable_options: dict[str, Any] = dict(options) if options else {} + + def _get_run_config() -> FunctionInvocationConfiguration: + base = dict(self.function_invocation_configuration) if self.function_invocation_configuration else {} + return cast("FunctionInvocationConfiguration", base) + execute_function_calls = partial( _execute_function_calls, custom_args=additional_function_arguments, - config=self.function_invocation_configuration, + config_provider=_get_run_config, invocation_session=invocation_session, approval_session_is_authoritative=approval_session_is_authoritative, middleware_pipeline=function_middleware_pipeline, host_payload_budget=host_payload_budget, ) - # Give the loop private mutable options and one shared run-local tool list for progressive tool changes. - # Make options mutable so we can update conversation_id during function invocation loop - mutable_options: dict[str, Any] = dict(options) if options else {} # Remove additional_function_arguments from options passed to underlying chat client # It's for tool invocation only and not recognized by chat service APIs mutable_options.pop("additional_function_arguments", None) diff --git a/python/packages/core/tests/core/test_tools.py b/python/packages/core/tests/core/test_tools.py index 89aa73486cc..7aa84a285ad 100644 --- a/python/packages/core/tests/core/test_tools.py +++ b/python/packages/core/tests/core/test_tools.py @@ -1,5 +1,6 @@ # Copyright (c) Microsoft. All rights reserved. import asyncio +import contextvars import logging import threading from typing import Annotated, Any, Literal, get_args, get_origin @@ -21,6 +22,7 @@ _auto_invoke_function, _parse_annotation, _parse_inputs, + _try_execute_function_call_groups, normalize_function_invocation_configuration, ) from agent_framework.observability import OtelAttr @@ -1754,3 +1756,144 @@ def test_skip_parsing_is_singleton() -> None: # endregion + + +async def test_try_execute_function_call_groups_sequential_config(): + """When allow_concurrent_invocation is False, ALL tools run one-by-one.""" + execution_order: list[str] = [] + + @tool() + async def tool_a(): + execution_order.append("a_start") + await asyncio.sleep(0.03) + execution_order.append("a_end") + return "a" + + @tool() + async def tool_b(): + execution_order.append("b_start") + await asyncio.sleep(0.01) + execution_order.append("b_end") + return "b" + + call_a = Content.from_function_call(call_id="1", name="tool_a", arguments="{}") + call_b = Content.from_function_call(call_id="2", name="tool_b", arguments="{}") + config = normalize_function_invocation_configuration({"allow_concurrent_invocation": False}) + results, should_terminate = await _try_execute_function_call_groups( + custom_args={}, + function_calls=[call_a, call_b], + tools=[tool_a, tool_b], + config=config, + ) + assert not should_terminate + assert execution_order == ["a_start", "a_end", "b_start", "b_end"] + + +async def test_sequential_execution_isolates_contextvars() -> None: + """ContextVar mutations in one tool should not leak to subsequent tools in sequential mode. + + Regression test requested by reviewer: ensures that `asyncio.create_task` is used + in the sequential execution path to isolate ContextVar mutations between tool calls. + """ + test_var: contextvars.ContextVar[str | None] = contextvars.ContextVar("test_var", default=None) + + @tool() + async def tool_a() -> str: + test_var.set("polluted_by_a") + return "a" + + @tool() + async def tool_b() -> str: + return test_var.get() or "None" + + call_a = Content.from_function_call(call_id="1", name="tool_a", arguments="{}") + call_b = Content.from_function_call(call_id="2", name="tool_b", arguments="{}") + + config = normalize_function_invocation_configuration({"allow_concurrent_invocation": False}) + results, should_terminate = await _try_execute_function_call_groups( + custom_args={}, + function_calls=[call_a, call_b], + tools=[tool_a, tool_b], + config=config, + ) + + assert not should_terminate + assert results[1][0].result == "None" + + +async def test_resolve_approval_responses_hold_later_approval_calls_in_sequential_mode() -> None: + """if [write[0], read(1), notify(2)] is requested, and `write` is pending approval while + `notify` is approved, `notify` must be held back. when `write` is later approved, + `notify` must be restored and executed. + """ + + from agent_framework._sessions import AgentSession + from agent_framework._tools import ( + _PENDING_APPROVAL_REQUESTS_KEY, + _TOOL_APPROVAL_STATE_KEY, + _FunctionExecutionBatch, + _resolve_approval_responses, + ) + from agent_framework._types import Message + + session = AgentSession() + write_call = Content.from_function_call(call_id="1", name="write_file", arguments="{}") + write_call.id = "1" + write_approval = Content.from_function_approval_request(id="1", function_call=write_call) + write_approval.additional_properties = {"original_index": 0, "batch_id": "batch_1"} + + notify_call = Content.from_function_call(call_id="3", name="notify_user", arguments="{}") + notify_call.id = "3" + notify_approval = Content.from_function_approval_request(id="3", function_call=notify_call) + notify_approval.additional_properties = {"original_index": 2, "batch_id": "batch_1"} + + session.state[_TOOL_APPROVAL_STATE_KEY] = { + _PENDING_APPROVAL_REQUESTS_KEY: [write_approval.to_dict(), notify_approval.to_dict()] + } + + notify_response = Content.from_function_approval_response(approved=True, id="3", function_call=notify_call) + notify_response.additional_properties = {"original_index": 2, "batch_id": "batch_1"} + + prepared_messages = [Message(role="user", contents=[notify_response])] + + executed_calls: list[Content] = [] + + async def mock_execute(*, function_calls, options): + executed_calls.extend(function_calls) + return _FunctionExecutionBatch(result_groups=[]) + + result = await _resolve_approval_responses( + prepared_messages=prepared_messages, + options={}, + errors_in_a_row=0, + max_errors=3, + execute_function_calls=mock_execute, + invocation_session=session, + allow_concurrent_invocation=False, + ) + + assert len(executed_calls) == 0, "notify_user executed before write_file was approved" + assert result.action == "return", "should return to wait for earlier approvals" + + # now user approved write (retry) + write_response = Content.from_function_approval_response(approved=True, id="1", function_call=write_call) + write_response.additional_properties = {"original_index": 0, "batch_id": "batch_1", "_approval_request_id": "1"} + + prepared_messages_retry = [Message(role="user", contents=[write_response])] + await _resolve_approval_responses( + prepared_messages=prepared_messages_retry, + options={}, + errors_in_a_row=0, + max_errors=3, + execute_function_calls=mock_execute, + invocation_session=session, + allow_concurrent_invocation=False, + ) + assert len(executed_calls) == 2, "write_file and notify_user should have executed on retry" + assert executed_calls[0].function_call is not None + assert executed_calls[0].function_call.name == "write_file" + assert executed_calls[1].function_call is not None + assert executed_calls[1].function_call.name == "notify_user" + + +# endregion