From bd7f2e065ef059dc57338ff5bc6ad41eddcf6168 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Thu, 17 Sep 2026 11:34:58 +0200 Subject: [PATCH 01/10] Python: add sequential function invocation option --- .../specs/004-python-function-calling-loop.md | 6 ++ .../packages/core/agent_framework/_tools.py | 65 +++++++++---- python/packages/core/tests/core/test_tools.py | 95 +++++++++++++++++++ 3 files changed, 145 insertions(+), 21 deletions(-) diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index bfbe4812299..b1eba25cc38 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -195,6 +195,12 @@ response is rejected. Correlation is scoped to the active mixed batch so complet remain unchanged. `ToolApprovalMiddleware` may resolve approval requests through standing or automatic policies, but it preserves non-approval user-input requests and does not split manual approvals away from their Host-owned siblings. +Fully executable calls run concurrently by default. When +`FunctionInvocationConfiguration["allow_concurrent_invocation"]` is `False`, the layer instead starts each call only +after the preceding call has completed, preserving model order. A middleware termination result prevents later calls +from starting; the layer emits synthetic skipped results for that unstarted suffix so every call remains paired with a +terminal result. + ### Reasoning-bound function-call groups Some hosted services bind reasoning content or an opaque reasoning signature to the function call that follows it. diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index 1caa4d82565..6e000901057 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -1696,6 +1696,9 @@ class FunctionInvocationConfiguration(TypedDict, total=False): sessions are always bounded, even if individual approval steps are slow. - ``max_consecutive_errors_per_request``: How many consecutive errors before abandoning the tool loop for this request. + - ``allow_concurrent_invocation``: Whether executable calls from one model + response may run concurrently. Set to ``False`` to execute them in model + order. Defaults to ``True``. - ``terminate_on_unknown_calls``: Whether to raise an error when the model requests a function that is not in the tool map. - ``additional_tools``: Extra tools available during execution but not @@ -1735,6 +1738,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( @@ -1749,6 +1753,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) @@ -2386,11 +2391,10 @@ async def _try_execute_function_call_groups( declaration_only_calls.append(_as_user_input_pause(function_call)) return [[function_call] for function_call in declaration_only_calls], False - # 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( + # Only a fully executable batch reaches this point. Each call gets its own + # task so context changes made by one tool cannot leak into another. + def create_execution_task(function_call: Content) -> asyncio.Task[tuple[list[Content], bool]]: + return contextvars.copy_context().run( asyncio.create_task, _execute_single_function_call( function_call, @@ -2403,22 +2407,41 @@ 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 + + execution_results: list[tuple[list[Content], bool]] = [] + if config.get("allow_concurrent_invocation", True): + execution_tasks = [create_execution_task(function_call) 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 + else: + for index, function_call in enumerate(function_calls): + result = await create_execution_task(function_call) + execution_results.append(result) + if result[1]: + for skipped_call in function_calls[index + 1 :]: + source_call = _underlying_function_call(skipped_call) + execution_results.append(( + [ + Content.from_function_result( + call_id=source_call.call_id, # type: ignore[arg-type] + result="Skipped: a prior tool call in this batch requested termination.", + ) + ], + False, + )) + break should_terminate = any(terminate for _, terminate in execution_results) return [result_contents for result_contents, _ in execution_results], should_terminate diff --git a/python/packages/core/tests/core/test_tools.py b/python/packages/core/tests/core/test_tools.py index 89aa73486cc..505ccaba86c 100644 --- a/python/packages/core/tests/core/test_tools.py +++ b/python/packages/core/tests/core/test_tools.py @@ -2,6 +2,7 @@ import asyncio import logging import threading +from contextvars import ContextVar from typing import Annotated, Any, Literal, get_args, get_origin from unittest.mock import Mock @@ -10,6 +11,7 @@ from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter from pydantic import BaseModel +import agent_framework._tools as tools_module from agent_framework import ( SKIP_PARSING, Content, @@ -21,6 +23,7 @@ _auto_invoke_function, _parse_annotation, _parse_inputs, + _try_execute_function_call_groups, normalize_function_invocation_configuration, ) from agent_framework.observability import OtelAttr @@ -28,6 +31,98 @@ # region FunctionTool and tool decorator tests +async def test_sequential_function_invocation_runs_calls_in_model_order() -> None: + execution_order: list[str] = [] + + @tool + async def first() -> str: + execution_order.append("first_start") + await asyncio.sleep(0) + execution_order.append("first_end") + return "first" + + @tool + async def second() -> str: + execution_order.append("second_start") + await asyncio.sleep(0) + execution_order.append("second_end") + return "second" + + result_groups, should_terminate = await _try_execute_function_call_groups( + custom_args={}, + function_calls=[ + Content.from_function_call(call_id="first", name="first", arguments={}), + Content.from_function_call(call_id="second", name="second", arguments={}), + ], + tools=[first, second], + config={"allow_concurrent_invocation": False}, + ) + + assert not should_terminate + assert execution_order == ["first_start", "first_end", "second_start", "second_end"] + assert [group[0].result for group in result_groups] == ["first", "second"] + + +async def test_sequential_function_invocation_isolates_context() -> None: + current_tool = ContextVar[str | None]("current_tool", default=None) + + @tool + async def first() -> str: + current_tool.set("first") + return "first" + + @tool + async def second() -> str: + return str(current_tool.get()) + + result_groups, _ = await _try_execute_function_call_groups( + custom_args={}, + function_calls=[ + Content.from_function_call(call_id="first", name="first", arguments={}), + Content.from_function_call(call_id="second", name="second", arguments={}), + ], + tools=[first, second], + config={"allow_concurrent_invocation": False}, + ) + + assert result_groups[1][0].result == "None" + + +def test_function_invocation_configuration_allows_concurrency_by_default() -> None: + config = normalize_function_invocation_configuration(None) + + assert config["allow_concurrent_invocation"] is True + + +async def test_sequential_function_invocation_skips_calls_after_termination( + monkeypatch: pytest.MonkeyPatch, +) -> None: + executed_calls: list[str | None] = [] + + async def execute(function_call: Content, **_: Any) -> tuple[list[Content], bool]: + executed_calls.append(function_call.call_id) + return [Content.from_function_result(call_id=function_call.call_id, result="done")], True # type: ignore[arg-type] + + monkeypatch.setattr(tools_module, "_execute_single_function_call", execute) + + result_groups, should_terminate = await _try_execute_function_call_groups( + custom_args={}, + function_calls=[ + Content.from_function_call(call_id="first", name="first", arguments={}), + Content.from_function_call(call_id="second", name="second", arguments={}), + ], + tools=[ + FunctionTool(name="first", func=lambda: "first"), + FunctionTool(name="second", func=lambda: "second"), + ], + config={"allow_concurrent_invocation": False}, + ) + + assert should_terminate + assert executed_calls == ["first"] + assert result_groups[1][0].result == "Skipped: a prior tool call in this batch requested termination." + + def test_tool_decorator(): """Test the tool decorator.""" From a1a00324da03f4a662003241b0b16638c9eb6d25 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Thu, 17 Sep 2026 12:03:25 +0200 Subject: [PATCH 02/10] Python: address sequential invocation review feedback --- .../specs/004-python-function-calling-loop.md | 6 +- .../packages/core/agent_framework/_tools.py | 65 +++++++++++++------ .../tests/core/test_harness_tool_approval.py | 5 ++ python/packages/core/tests/core/test_tools.py | 14 ++-- 4 files changed, 60 insertions(+), 30 deletions(-) diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index b1eba25cc38..6ba1ad5fe9b 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -197,9 +197,8 @@ it preserves non-approval user-input requests and does not split manual approval Fully executable calls run concurrently by default. When `FunctionInvocationConfiguration["allow_concurrent_invocation"]` is `False`, the layer instead starts each call only -after the preceding call has completed, preserving model order. A middleware termination result prevents later calls -from starting; the layer emits synthetic skipped results for that unstarted suffix so every call remains paired with a -terminal result. +after the preceding call has completed, preserving model order across approval pause and replay. Middleware termination +stops the loop after the accepted batch finishes; it does not skip later calls in that batch. ### Reasoning-bound function-call groups @@ -552,6 +551,7 @@ that manually replay messages own the equivalent rule: do not resend an approval | Reasoning-bound call | Finalized output retains reasoning, function call, function result, and final text. | `test_streaming_function_calling_response_includes_reasoning_and_tool_results` | | Calls across response messages | Every actionable call is executed once. | `test_base_client_executes_function_calls_across_multiple_response_messages` | | Parallel calls | Results retain the corresponding call ids and execution count. | `test_max_function_calls_limits_parallel_invocations`, `test_streaming_multiple_function_calls_parallel_execution` | +| Sequential calls | Calls execute one at a time in model order, including after approval replay; each call has isolated context, and middleware termination takes effect after the accepted batch finishes. | `packages/core/tests/core/test_tools.py::test_sequential_function_invocation_runs_calls_in_model_order`, `test_sequential_function_invocation_isolates_context`, `test_sequential_function_invocation_finishes_batch_after_termination`, `packages/core/tests/core/test_harness_tool_approval.py::test_mixed_batch_hides_already_approved_request_until_approval_replay` | | Informational-only call | The call is returned but not executed or approved. | `test_informational_only_function_call_is_not_invoked`, `test_informational_only_function_call_does_not_request_approval`, `test_streaming_informational_only_function_call_is_not_invoked` | | OpenAI hosted/local shell boundary | Hosted shell calls remain informational in streaming and non-streaming responses even when a local executor is configured; only valid explicit local-shell items or shell calls marked with a local environment can execute, local execution preserves its configured approval mode, stateless loops preserve the complete provider shell transcript or fail explicitly, and locally generated shell outputs are sent in every continuation mode. | `packages/openai/tests/openai/test_openai_chat_client.py::test_response_content_creation_with_shell_call_remains_hosted_with_local_tool`, `test_parse_chunk_from_openai_shell_call_done_remains_hosted`, `test_parse_chunk_from_openai_local_environment_shell_call_done_emits_command`, `test_mixed_shell_calls_only_invoke_explicit_local_shell_call`, `test_stateless_shell_transcript_without_provider_item_fails`, `test_prepare_messages_keeps_local_shell_output_under_storage`, `test_malformed_local_environment_shell_call_is_not_executable`, `test_response_content_creation_with_local_shell_call_maps_to_function_call`, `test_malformed_local_shell_call_is_not_executable`, `test_response_function_call_named_local_shell_is_informational`, `test_parse_chunk_function_call_named_local_shell_is_informational`, `test_local_shell_tool_requires_approval_before_function_loop_execution` | | Declaration-only call | The call is surfaced as user input and is not executed; streaming arguments appear once while finalized request metadata remains available. | `test_declaration_only_tool`, `test_streaming_declaration_only_tool_preserves_metadata_without_duplicate_arguments` | diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index 6e000901057..c42a795f093 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -117,6 +117,7 @@ def _has_authoritative_approval_session(invocation_session: AgentSession | None) _ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY: Final[str] = "already_approved_approval_request_groups" +_APPROVAL_REQUEST_ORDER_KEY: Final[str] = "approval_request_order" _PENDING_APPROVAL_REQUESTS_KEY: Final[str] = "pending_approval_requests" _PENDING_MIXED_PAUSE_BATCH_KEY: Final[str] = "pending_mixed_pause_batch" _APPROVAL_REQUEST_ID_KEY: Final[str] = "_approval_request_id" @@ -2343,6 +2344,7 @@ async def _try_execute_function_call_groups( logger.debug("Returning visible function_approval_request contents and storing already-approved requests") visible_requests: list[Content] = [] already_approved_requests: list[Content] = [] + approval_request_order: list[str] = [] pause_groups: list[list[Content]] = [] for function_call in function_calls: if function_call.type != "function_call": @@ -2359,6 +2361,8 @@ async def _try_execute_function_call_groups( id=function_call.id or function_call.call_id, # type: ignore[arg-type] function_call=function_call, ) + if approval_request.id is not None: + approval_request_order.append(approval_request.id) if tool_name is None: visible_requests.append(approval_request) pause_groups.append([approval_request]) @@ -2377,6 +2381,7 @@ async def _try_execute_function_call_groups( invocation_session, visible_requests, already_approved_requests, + approval_request_order=approval_request_order, ) _store_pending_approval_requests(invocation_session, visible_requests) _store_pending_mixed_pause_batch(invocation_session, pause_groups) @@ -2426,22 +2431,9 @@ def create_execution_task(function_call: Content) -> asyncio.Task[tuple[list[Con await asyncio.gather(*execution_tasks, return_exceptions=True) raise else: - for index, function_call in enumerate(function_calls): + for function_call in function_calls: result = await create_execution_task(function_call) execution_results.append(result) - if result[1]: - for skipped_call in function_calls[index + 1 :]: - source_call = _underlying_function_call(skipped_call) - execution_results.append(( - [ - Content.from_function_result( - call_id=source_call.call_id, # type: ignore[arg-type] - result="Skipped: a prior tool call in this batch requested termination.", - ) - ], - False, - )) - break should_terminate = any(terminate for _, terminate in execution_results) return [result_contents for result_contents, _ in execution_results], should_terminate @@ -2892,6 +2884,8 @@ def _store_already_approved_approval_requests( invocation_session: AgentSession | None, visible_approval_requests: Sequence[Content], already_approved_requests: Sequence[Content], + *, + approval_request_order: Sequence[str] | None = None, ) -> None: """Store hidden already-approved requests keyed by the visible approvals that resume the batch.""" if not already_approved_requests: @@ -2908,6 +2902,11 @@ def _store_already_approved_approval_requests( pending_groups.append({ "approval_request_ids": visible_ids, "approval_requests": [request.to_dict() for request in already_approved_requests], + _APPROVAL_REQUEST_ORDER_KEY: list(approval_request_order) + if approval_request_order is not None + else [ + request.id for request in (*visible_approval_requests, *already_approved_requests) if request.id is not None + ], }) state[_ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY] = pending_groups @@ -2915,19 +2914,20 @@ def _store_already_approved_approval_requests( def _pop_already_approved_approval_responses( invocation_session: AgentSession | None, approval_response_ids: set[str], -) -> list[Content]: +) -> tuple[list[Content], list[str]]: """Pop already-approved requests for the visible approval ids being answered.""" if not approval_response_ids: - return [] + return [], [] state = _get_tool_approval_state(invocation_session) if state is None: - return [] + return [], [] raw_groups = state.get(_ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY, []) if not isinstance(raw_groups, list): - return [] + return [], [] typed_groups = cast(list[Any], raw_groups) responses: list[Content] = [] + approval_request_order: list[str] = [] remaining_groups: list[Any] = [] for raw_group in typed_groups: if not isinstance(raw_group, Mapping): @@ -2938,6 +2938,9 @@ def _pop_already_approved_approval_responses( if group_ids.isdisjoint(approval_response_ids): remaining_groups.append(raw_group) continue + raw_order = group.get(_APPROVAL_REQUEST_ORDER_KEY) + if isinstance(raw_order, list): + approval_request_order.extend(str(item) for item in cast(list[Any], raw_order)) raw_requests = group.get("approval_requests") if not isinstance(raw_requests, list): continue @@ -2950,7 +2953,7 @@ def _pop_already_approved_approval_responses( state[_ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY] = remaining_groups else: state.pop(_ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY, None) - return responses + return responses, approval_request_order def _store_pending_mixed_pause_batch( @@ -4088,11 +4091,31 @@ async def _resolve_approval_responses( if content.type == "function_approval_response" and content.id } - if already_approved_responses := _pop_already_approved_approval_responses( + already_approved_responses, approval_request_order = _pop_already_approved_approval_responses( invocation_session, explicit_approval_response_ids, - ): + ) + if already_approved_responses: prepared_messages.append(Message(role="user", contents=already_approved_responses)) + responses_by_id: dict[str, Content] = {} + for message in prepared_messages: + retained_contents: list[Content] = [] + for content in message.contents: + if ( + content.type == "function_approval_response" + and content.id is not None + and content.id in approval_request_order + ): + responses_by_id[content.id] = content + else: + retained_contents.append(content) + message.contents = retained_contents + prepared_messages[:] = [message for message in prepared_messages if message.contents] + ordered_responses = [ + responses_by_id[request_id] for request_id in approval_request_order if request_id in responses_by_id + ] + if ordered_responses: + prepared_messages.append(Message(role="user", contents=ordered_responses)) # 2. With no new decision, hide any still-pending batch from model input while keeping it resumable in history. if not ( diff --git a/python/packages/core/tests/core/test_harness_tool_approval.py b/python/packages/core/tests/core/test_harness_tool_approval.py index 95f65e5a467..530686f4826 100644 --- a/python/packages/core/tests/core/test_harness_tool_approval.py +++ b/python/packages/core/tests/core/test_harness_tool_approval.py @@ -1369,20 +1369,24 @@ async def test_mixed_batch_hides_already_approved_request_until_approval_replay( """Mixed batches should only show real approval requests when a session can store hidden requests.""" no_approval_calls = 0 approval_calls = 0 + execution_order: list[str] = [] @tool(name="lookup_work_items", approval_mode="never_require") def lookup_work_items(query: str) -> str: nonlocal no_approval_calls no_approval_calls += 1 + execution_order.append("lookup_work_items") return f"found {query}" @tool(name="add_comment", approval_mode="always_require") def add_comment(comment: str) -> str: nonlocal approval_calls approval_calls += 1 + execution_order.append("add_comment") return f"added {comment}" agent = Agent(client=chat_client_base, tools=[lookup_work_items, add_comment]) + chat_client_base.function_invocation_configuration["allow_concurrent_invocation"] = False session = AgentSession(session_id="approval-session") chat_client_base.run_responses = [ ChatResponse( @@ -1417,6 +1421,7 @@ def add_comment(comment: str) -> str: assert second_response.text == "complete" assert no_approval_calls == 1 assert approval_calls == 1 + assert execution_order == ["lookup_work_items", "add_comment"] async def test_mixed_batch_accepts_restored_tool_approval_state( diff --git a/python/packages/core/tests/core/test_tools.py b/python/packages/core/tests/core/test_tools.py index 505ccaba86c..4b0eb1c81a9 100644 --- a/python/packages/core/tests/core/test_tools.py +++ b/python/packages/core/tests/core/test_tools.py @@ -94,14 +94,16 @@ def test_function_invocation_configuration_allows_concurrency_by_default() -> No assert config["allow_concurrent_invocation"] is True -async def test_sequential_function_invocation_skips_calls_after_termination( +async def test_sequential_function_invocation_finishes_batch_after_termination( monkeypatch: pytest.MonkeyPatch, ) -> None: - executed_calls: list[str | None] = [] + executed_calls: list[str] = [] async def execute(function_call: Content, **_: Any) -> tuple[list[Content], bool]: - executed_calls.append(function_call.call_id) - return [Content.from_function_result(call_id=function_call.call_id, result="done")], True # type: ignore[arg-type] + call_id = function_call.call_id + assert call_id is not None + executed_calls.append(call_id) + return [Content.from_function_result(call_id=call_id, result="done")], call_id == "first" monkeypatch.setattr(tools_module, "_execute_single_function_call", execute) @@ -119,8 +121,8 @@ async def execute(function_call: Content, **_: Any) -> tuple[list[Content], bool ) assert should_terminate - assert executed_calls == ["first"] - assert result_groups[1][0].result == "Skipped: a prior tool call in this batch requested termination." + assert executed_calls == ["first", "second"] + assert [group[0].result for group in result_groups] == ["done", "done"] def test_tool_decorator(): From d16c30759f163d0f82c732b6b9892fcba01f9bf9 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Thu, 17 Sep 2026 13:22:14 +0200 Subject: [PATCH 03/10] Python: preserve order for approval-only batches --- .../specs/004-python-function-calling-loop.md | 2 +- .../packages/core/agent_framework/_tools.py | 7 ++- .../tests/core/test_harness_tool_approval.py | 51 +++++++++++++++++++ 3 files changed, 55 insertions(+), 5 deletions(-) diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index e5d55c1d857..b4ff05adb0a 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -554,7 +554,7 @@ that manually replay messages own the equivalent rule: do not resend an approval | Reasoning-bound call | Finalized output retains reasoning, function call, function result, and final text. | `test_streaming_function_calling_response_includes_reasoning_and_tool_results` | | Calls across response messages | Every actionable call is executed once. | `test_base_client_executes_function_calls_across_multiple_response_messages` | | Parallel calls | Results retain the corresponding call ids and execution count. | `test_max_function_calls_limits_parallel_invocations`, `test_streaming_multiple_function_calls_parallel_execution` | -| Sequential calls | Calls execute one at a time in model order, including after approval replay; each call has isolated context, and middleware termination takes effect after the accepted batch finishes. | `packages/core/tests/core/test_tools.py::test_sequential_function_invocation_runs_calls_in_model_order`, `test_sequential_function_invocation_isolates_context`, `test_sequential_function_invocation_finishes_batch_after_termination`, `packages/core/tests/core/test_harness_tool_approval.py::test_mixed_batch_hides_already_approved_request_until_approval_replay` | +| Sequential calls | Calls execute one at a time in model order, including after approval replay; each call has isolated context, and middleware termination takes effect after the accepted batch finishes. | `packages/core/tests/core/test_tools.py::test_sequential_function_invocation_runs_calls_in_model_order`, `test_sequential_function_invocation_isolates_context`, `test_sequential_function_invocation_finishes_batch_after_termination`, `packages/core/tests/core/test_harness_tool_approval.py::test_mixed_batch_hides_already_approved_request_until_approval_replay`, `test_sequential_approval_replay_preserves_model_order_when_responses_are_reversed` | | Informational-only call | The call is returned but not executed or approved. | `test_informational_only_function_call_is_not_invoked`, `test_informational_only_function_call_does_not_request_approval`, `test_streaming_informational_only_function_call_is_not_invoked` | | OpenAI hosted/local shell boundary | Hosted shell calls remain informational in streaming and non-streaming responses even when a local executor is configured; only valid explicit local-shell items or shell calls marked with a local environment can execute, local execution preserves its configured approval mode, stateless loops preserve the complete provider shell transcript or fail explicitly, and locally generated shell outputs are sent in every continuation mode. | `packages/openai/tests/openai/test_openai_chat_client.py::test_response_content_creation_with_shell_call_remains_hosted_with_local_tool`, `test_parse_chunk_from_openai_shell_call_done_remains_hosted`, `test_parse_chunk_from_openai_local_environment_shell_call_done_emits_command`, `test_mixed_shell_calls_only_invoke_explicit_local_shell_call`, `test_stateless_shell_transcript_without_provider_item_fails`, `test_prepare_messages_keeps_local_shell_output_under_storage`, `test_malformed_local_environment_shell_call_is_not_executable`, `test_response_content_creation_with_local_shell_call_maps_to_function_call`, `test_malformed_local_shell_call_is_not_executable`, `test_response_function_call_named_local_shell_is_informational`, `test_parse_chunk_function_call_named_local_shell_is_informational`, `test_local_shell_tool_requires_approval_before_function_loop_execution` | | Declaration-only call | The call is surfaced as user input and is not executed; streaming arguments appear once while finalized request metadata remains available. | `test_declaration_only_tool`, `test_streaming_declaration_only_tool_preserves_metadata_without_duplicate_arguments` | diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index 882e3126d9b..1217ed3daf4 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -2890,9 +2890,7 @@ def _store_already_approved_approval_requests( *, approval_request_order: Sequence[str] | None = None, ) -> None: - """Store hidden already-approved requests keyed by the visible approvals that resume the batch.""" - if not already_approved_requests: - return + """Store approval order and hidden safe requests keyed by the visible approvals that resume the batch.""" state = _get_tool_approval_state(invocation_session) if state is None: return @@ -2918,7 +2916,7 @@ def _pop_already_approved_approval_responses( invocation_session: AgentSession | None, approval_response_ids: set[str], ) -> tuple[list[Content], list[str]]: - """Pop already-approved requests for the visible approval ids being answered.""" + """Pop the stored order and hidden safe requests for the approval ids being answered.""" if not approval_response_ids: return [], [] state = _get_tool_approval_state(invocation_session) @@ -4097,6 +4095,7 @@ async def _resolve_approval_responses( ) if already_approved_responses: prepared_messages.append(Message(role="user", contents=already_approved_responses)) + if approval_request_order: responses_by_id: dict[str, Content] = {} for message in prepared_messages: retained_contents: list[Content] = [] diff --git a/python/packages/core/tests/core/test_harness_tool_approval.py b/python/packages/core/tests/core/test_harness_tool_approval.py index e3b90b2c2b8..4e5dd99f1c5 100644 --- a/python/packages/core/tests/core/test_harness_tool_approval.py +++ b/python/packages/core/tests/core/test_harness_tool_approval.py @@ -1557,6 +1557,57 @@ def add_comment(comment: str) -> str: assert execution_order == ["lookup_work_items", "add_comment"] +async def test_sequential_approval_replay_preserves_model_order_when_responses_are_reversed( + chat_client_base: MockBaseChatClient, +) -> None: + """Sequential approval replay should follow model order rather than caller response order.""" + execution_order: list[str] = [] + + @tool(name="first_write", approval_mode="always_require") + def first_write() -> str: + execution_order.append("first_write") + return "first" + + @tool(name="second_write", approval_mode="always_require") + def second_write() -> str: + execution_order.append("second_write") + return "second" + + agent = Agent(client=chat_client_base, tools=[first_write, second_write]) + chat_client_base.function_invocation_configuration["allow_concurrent_invocation"] = False + session = AgentSession(session_id="reversed-approval-order") + chat_client_base.run_responses = [ + ChatResponse( + messages=Message( + role="assistant", + contents=[ + Content.from_function_call(call_id="call_first", name="first_write", arguments="{}"), + Content.from_function_call(call_id="call_second", name="second_write", arguments="{}"), + ], + ) + ) + ] + + first_response = await agent.run("write in order", session=session) + requests = _approval_requests(first_response.messages) + assert [_function_call(request).name for request in requests] == ["first_write", "second_write"] + + chat_client_base.run_responses = [ChatResponse(messages=Message(role="assistant", contents=["complete"]))] + final_response = await agent.run( + Message( + role="user", + contents=[ + requests[1].to_function_approval_response(approved=True), + requests[0].to_function_approval_response(approved=True), + ], + ), + session=session, + ) + + assert final_response.text == "complete" + assert execution_order == ["first_write", "second_write"] + + async def test_mixed_batch_accepts_restored_tool_approval_state( chat_client_base: MockBaseChatClient, ) -> None: From 2e725bf20c0983e3c42b5a2a023fb08eeb16d35d Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Thu, 17 Sep 2026 13:37:21 +0200 Subject: [PATCH 04/10] Python: preserve mixed approval replay order --- .../specs/004-python-function-calling-loop.md | 2 +- .../packages/core/agent_framework/_tools.py | 102 ++++++++++++------ .../core/test_function_invocation_logic.py | 83 ++++++++++++++ 3 files changed, 156 insertions(+), 31 deletions(-) diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index b4ff05adb0a..ee3c44c6988 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -554,7 +554,7 @@ that manually replay messages own the equivalent rule: do not resend an approval | Reasoning-bound call | Finalized output retains reasoning, function call, function result, and final text. | `test_streaming_function_calling_response_includes_reasoning_and_tool_results` | | Calls across response messages | Every actionable call is executed once. | `test_base_client_executes_function_calls_across_multiple_response_messages` | | Parallel calls | Results retain the corresponding call ids and execution count. | `test_max_function_calls_limits_parallel_invocations`, `test_streaming_multiple_function_calls_parallel_execution` | -| Sequential calls | Calls execute one at a time in model order, including after approval replay; each call has isolated context, and middleware termination takes effect after the accepted batch finishes. | `packages/core/tests/core/test_tools.py::test_sequential_function_invocation_runs_calls_in_model_order`, `test_sequential_function_invocation_isolates_context`, `test_sequential_function_invocation_finishes_batch_after_termination`, `packages/core/tests/core/test_harness_tool_approval.py::test_mixed_batch_hides_already_approved_request_until_approval_replay`, `test_sequential_approval_replay_preserves_model_order_when_responses_are_reversed` | +| Sequential calls | Calls execute one at a time in model order, including after approval replay; mixed local/Host-owned siblings retain that order, each call has isolated context, and middleware termination takes effect after the accepted batch finishes. | `packages/core/tests/core/test_tools.py::test_sequential_function_invocation_runs_calls_in_model_order`, `test_sequential_function_invocation_isolates_context`, `test_sequential_function_invocation_finishes_batch_after_termination`, `packages/core/tests/core/test_harness_tool_approval.py::test_mixed_batch_hides_already_approved_request_until_approval_replay`, `test_sequential_approval_replay_preserves_model_order_when_responses_are_reversed`, `packages/core/tests/core/test_function_invocation_logic.py::test_sequential_mixed_batch_preserves_approval_and_host_model_order` | | Informational-only call | The call is returned but not executed or approved. | `test_informational_only_function_call_is_not_invoked`, `test_informational_only_function_call_does_not_request_approval`, `test_streaming_informational_only_function_call_is_not_invoked` | | OpenAI hosted/local shell boundary | Hosted shell calls remain informational in streaming and non-streaming responses even when a local executor is configured; only valid explicit local-shell items or shell calls marked with a local environment can execute, local execution preserves its configured approval mode, stateless loops preserve the complete provider shell transcript or fail explicitly, and locally generated shell outputs are sent in every continuation mode. | `packages/openai/tests/openai/test_openai_chat_client.py::test_response_content_creation_with_shell_call_remains_hosted_with_local_tool`, `test_parse_chunk_from_openai_shell_call_done_remains_hosted`, `test_parse_chunk_from_openai_local_environment_shell_call_done_emits_command`, `test_mixed_shell_calls_only_invoke_explicit_local_shell_call`, `test_stateless_shell_transcript_without_provider_item_fails`, `test_prepare_messages_keeps_local_shell_output_under_storage`, `test_malformed_local_environment_shell_call_is_not_executable`, `test_response_content_creation_with_local_shell_call_maps_to_function_call`, `test_malformed_local_shell_call_is_not_executable`, `test_response_function_call_named_local_shell_is_informational`, `test_parse_chunk_function_call_named_local_shell_is_informational`, `test_local_shell_tool_requires_approval_before_function_loop_execution` | | Declaration-only call | The call is surfaced as user input and is not executed; streaming arguments appear once while finalized request metadata remains available. | `test_declaration_only_tool`, `test_streaming_declaration_only_tool_preserves_metadata_without_duplicate_arguments` | diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index 1217ed3daf4..9541ab5f6ac 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -114,7 +114,7 @@ def _has_authoritative_approval_session(invocation_session: AgentSession | None) _ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY: Final[str] = "already_approved_approval_request_groups" -_APPROVAL_REQUEST_ORDER_KEY: Final[str] = "approval_request_order" +_FUNCTION_CALL_ORDER_KEY: Final[str] = "function_call_order" _PENDING_APPROVAL_REQUESTS_KEY: Final[str] = "pending_approval_requests" _PENDING_MIXED_PAUSE_BATCH_KEY: Final[str] = "pending_mixed_pause_batch" _APPROVAL_REQUEST_ID_KEY: Final[str] = "_approval_request_id" @@ -2345,7 +2345,11 @@ async def _try_execute_function_call_groups( logger.debug("Returning visible function_approval_request contents and storing already-approved requests") visible_requests: list[Content] = [] already_approved_requests: list[Content] = [] - approval_request_order: list[str] = [] + function_call_order = [ + {"id": function_call.id, "call_id": function_call.call_id} + for function_call in function_calls + if function_call.type == "function_call" + ] pause_groups: list[list[Content]] = [] for function_call in function_calls: if function_call.type != "function_call": @@ -2362,8 +2366,6 @@ async def _try_execute_function_call_groups( id=function_call.id or function_call.call_id, # type: ignore[arg-type] function_call=function_call, ) - if approval_request.id is not None: - approval_request_order.append(approval_request.id) if tool_name is None: visible_requests.append(approval_request) pause_groups.append([approval_request]) @@ -2382,7 +2384,7 @@ async def _try_execute_function_call_groups( approval_session, visible_requests, already_approved_requests, - approval_request_order=approval_request_order, + function_call_order=function_call_order, ) _store_pending_approval_requests(approval_session, visible_requests) _store_pending_mixed_pause_batch(approval_session, pause_groups) @@ -2888,7 +2890,7 @@ def _store_already_approved_approval_requests( visible_approval_requests: Sequence[Content], already_approved_requests: Sequence[Content], *, - approval_request_order: Sequence[str] | None = None, + function_call_order: Sequence[Mapping[str, str | None]] | None = None, ) -> None: """Store approval order and hidden safe requests keyed by the visible approvals that resume the batch.""" state = _get_tool_approval_state(invocation_session) @@ -2903,10 +2905,12 @@ def _store_already_approved_approval_requests( pending_groups.append({ "approval_request_ids": visible_ids, "approval_requests": [request.to_dict() for request in already_approved_requests], - _APPROVAL_REQUEST_ORDER_KEY: list(approval_request_order) - if approval_request_order is not None + _FUNCTION_CALL_ORDER_KEY: [dict(item) for item in function_call_order] + if function_call_order is not None else [ - request.id for request in (*visible_approval_requests, *already_approved_requests) if request.id is not None + {"id": request.id, "call_id": request.function_call.call_id} + for request in (*visible_approval_requests, *already_approved_requests) + if request.function_call is not None ], }) state[_ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY] = pending_groups @@ -2915,7 +2919,7 @@ def _store_already_approved_approval_requests( def _pop_already_approved_approval_responses( invocation_session: AgentSession | None, approval_response_ids: set[str], -) -> tuple[list[Content], list[str]]: +) -> tuple[list[Content], list[dict[str, str | None]]]: """Pop the stored order and hidden safe requests for the approval ids being answered.""" if not approval_response_ids: return [], [] @@ -2928,7 +2932,7 @@ def _pop_already_approved_approval_responses( typed_groups = cast(list[Any], raw_groups) responses: list[Content] = [] - approval_request_order: list[str] = [] + function_call_order: list[dict[str, str | None]] = [] remaining_groups: list[Any] = [] for raw_group in typed_groups: if not isinstance(raw_group, Mapping): @@ -2939,9 +2943,18 @@ def _pop_already_approved_approval_responses( if group_ids.isdisjoint(approval_response_ids): remaining_groups.append(raw_group) continue - raw_order = group.get(_APPROVAL_REQUEST_ORDER_KEY) + raw_order = group.get(_FUNCTION_CALL_ORDER_KEY) if isinstance(raw_order, list): - approval_request_order.extend(str(item) for item in cast(list[Any], raw_order)) + for raw_item in cast(list[Any], raw_order): + if not isinstance(raw_item, Mapping): + continue + item = cast(Mapping[str, Any], raw_item) + item_id = item.get("id") + call_id = item.get("call_id") + function_call_order.append({ + "id": str(item_id) if item_id is not None else None, + "call_id": str(call_id) if call_id is not None else None, + }) raw_requests = group.get("approval_requests") if not isinstance(raw_requests, list): continue @@ -2954,7 +2967,7 @@ def _pop_already_approved_approval_responses( state[_ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY] = remaining_groups else: state.pop(_ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY, None) - return responses, approval_request_order + return responses, function_call_order def _store_pending_mixed_pause_batch( @@ -4089,31 +4102,60 @@ async def _resolve_approval_responses( if content.type == "function_approval_response" and content.id } - already_approved_responses, approval_request_order = _pop_already_approved_approval_responses( + already_approved_responses, function_call_order = _pop_already_approved_approval_responses( approval_session, explicit_approval_response_ids, ) if already_approved_responses: prepared_messages.append(Message(role="user", contents=already_approved_responses)) - if approval_request_order: - responses_by_id: dict[str, Content] = {} + if function_call_order: + ordered_ids = {item["id"] for item in function_call_order if item["id"] is not None} + ordered_call_ids = {item["call_id"] for item in function_call_order if item["call_id"] is not None} + batch_responses: list[Content] = [] for message in prepared_messages: - retained_contents: list[Content] = [] for content in message.contents: + function_call = content.function_call if content.type == "function_approval_response" else None + call_id = function_call.call_id if function_call is not None else content.call_id if ( content.type == "function_approval_response" - and content.id is not None - and content.id in approval_request_order - ): - responses_by_id[content.id] = content - else: - retained_contents.append(content) - message.contents = retained_contents - prepared_messages[:] = [message for message in prepared_messages if message.contents] - ordered_responses = [ - responses_by_id[request_id] for request_id in approval_request_order if request_id in responses_by_id - ] - if ordered_responses: + and (content.id in ordered_ids or call_id in ordered_call_ids) + ) or id(content) in host_result_ids: + batch_responses.append(content) + + remaining_responses = list(batch_responses) + ordered_responses: list[Content] = [] + for order_item in function_call_order: + matching_index = next( + ( + index + for index, response in enumerate(remaining_responses) + if order_item["id"] is not None and response.id == order_item["id"] + ), + None, + ) + if matching_index is None: + matching_index = next( + ( + index + for index, response in enumerate(remaining_responses) + if order_item["call_id"] is not None + and ( + response.function_call.call_id + if response.type == "function_approval_response" and response.function_call is not None + else response.call_id + ) + == order_item["call_id"] + ), + None, + ) + if matching_index is not None: + ordered_responses.append(remaining_responses.pop(matching_index)) + + ordered_response_ids = {id(response) for response in ordered_responses} + if ordered_response_ids: + for message in prepared_messages: + message.contents = [content for content in message.contents if id(content) not in ordered_response_ids] + prepared_messages[:] = [message for message in prepared_messages if message.contents] prepared_messages.append(Message(role="user", contents=ordered_responses)) # 2. With no new decision, hide any still-pending batch from model input while keeping it resumable in history. 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 9dea58bf4c2..c1f1586b0a6 100644 --- a/python/packages/core/tests/core/test_function_invocation_logic.py +++ b/python/packages/core/tests/core/test_function_invocation_logic.py @@ -3463,6 +3463,89 @@ def approval_func(value: str) -> str: assert approval_arguments == ["expected"] +async def test_sequential_mixed_batch_preserves_approval_and_host_model_order( + chat_client_base: SupportsChatGetResponse, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Sequential replay should not move an approval result behind its Host-owned sibling.""" + from agent_framework import FunctionTool + + @tool(name="approval_func", approval_mode="always_require") + def approval_func() -> str: + return "approved" + + host_func = FunctionTool(name="host_func", func=None, description="Handled by the caller") + agent = Agent(client=chat_client_base, tools=[approval_func, host_func]) + chat_client_base.function_invocation_configuration["allow_concurrent_invocation"] = False # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + continuation_result_orders: list[list[str | None]] = [] + original_inner_get_response = chat_client_base._inner_get_response # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + + def capture_inner_get_response(**kwargs: Any) -> Any: + continuation_result_orders.append([ + content.call_id + for message in kwargs["messages"] + for content in message.contents + if content.type == "function_result" + ]) + return original_inner_get_response(**kwargs) + + monkeypatch.setattr(chat_client_base, "_inner_get_response", capture_inner_get_response) + session = AgentSession() + 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="approval", + name="approval_func", + arguments={}, + id="approval-occurrence", + ), + Content.from_function_call( + call_id="host", + name="host_func", + arguments={}, + id="host-occurrence", + ), + ], + ) + ), + ChatResponse(messages=Message(role="assistant", contents=["done"])), + ] + + first_response = await agent.run("run both", session=session) + approval_request = next( + content + for message in first_response.messages + for content in message.contents + if content.type == "function_approval_request" + ) + host_request = next( + content + for message in first_response.messages + for content in message.contents + if content.type == "function_call" and content.user_input_request + ) + assert host_request.call_id is not None + host_result = Content.from_function_result(call_id=host_request.call_id, result="host result") + host_result.id = host_request.id + + final_response = await agent.run( + Message( + role="user", + contents=[ + approval_request.to_function_approval_response(approved=True), + host_result, + ], + ), + session=session, + ) + + assert final_response.text == "done" + assert continuation_result_orders[-1] == ["approval", "host"] + + @pytest.mark.parametrize("include_approval_response", [False, True], ids=["zero-responses", "approval-only"]) async def test_stateless_split_mixed_batch_rejects_incomplete_replay_before_execution( chat_client_base: SupportsChatGetResponse, From cb49f7827c23070fc0a7c79c0d5d850a7578073f Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Thu, 17 Sep 2026 13:48:47 +0200 Subject: [PATCH 05/10] fix: deduplicate ordered approval replay responses --- python/packages/core/agent_framework/_tools.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index 9541ab5f6ac..481dc8e11ba 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -4151,10 +4151,10 @@ async def _resolve_approval_responses( if matching_index is not None: ordered_responses.append(remaining_responses.pop(matching_index)) - ordered_response_ids = {id(response) for response in ordered_responses} - if ordered_response_ids: + batch_response_ids = {id(response) for response in batch_responses} + if batch_response_ids: for message in prepared_messages: - message.contents = [content for content in message.contents if id(content) not in ordered_response_ids] + message.contents = [content for content in message.contents if id(content) not in batch_response_ids] prepared_messages[:] = [message for message in prepared_messages if message.contents] prepared_messages.append(Message(role="user", contents=ordered_responses)) From 458b94a23a77c283dedf5892a6eab0b3df6d1df9 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Thu, 17 Sep 2026 14:06:33 +0200 Subject: [PATCH 06/10] fix: preserve stateless approval replay order --- .../packages/core/agent_framework/_tools.py | 39 ++++++++++++ .../core/test_function_invocation_logic.py | 62 +++++++++++++++++++ 2 files changed, 101 insertions(+) diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index 481dc8e11ba..2905456117e 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -3372,6 +3372,39 @@ def _collect_unanswered_approval_requests(messages: Sequence[Message]) -> list[C return list(unanswered_by_id.values()) +def _derive_stateless_approval_response_order( + messages: Sequence[Message], + approval_responses: Sequence[Content], +) -> list[dict[str, str | None]]: + """Recover approval order from unresolved requests when no session state exists.""" + response_content_ids = {id(response) for response in approval_responses} + history_without_responses: list[Message] = [] + for message in messages: + copied_message = copy.copy(message) + copied_message.contents = [content for content in message.contents if id(content) not in response_content_ids] + history_without_responses.append(copied_message) + + response_ids = {response.id for response in approval_responses if response.id is not None} + response_call_ids = { + response.function_call.call_id + for response in approval_responses + if response.function_call is not None and response.function_call.call_id is not None + } + order: list[dict[str, str | None]] = [] + for request in _collect_unanswered_approval_requests(history_without_responses): + function_call = request.function_call + if function_call is None: + continue + if ( + request.id not in response_ids + and function_call.id not in response_ids + and function_call.call_id not in response_call_ids + ): + continue + order.append({"id": function_call.id or request.id, "call_id": function_call.call_id}) + return order + + def _remove_unanswered_approval_batches_from_model_input(messages: list[Message]) -> None: pending_requests = _collect_unanswered_approval_requests(messages) if not pending_requests: @@ -4108,6 +4141,12 @@ async def _resolve_approval_responses( ) if already_approved_responses: prepared_messages.append(Message(role="user", contents=already_approved_responses)) + if not function_call_order and approval_session is None and not host_result_ids: + pending_stateless_responses = _collect_approval_responses(prepared_messages) + function_call_order = _derive_stateless_approval_response_order( + prepared_messages, + list(pending_stateless_responses.values()), + ) if function_call_order: ordered_ids = {item["id"] for item in function_call_order if item["id"] is not None} ordered_call_ids = {item["call_id"] for item in function_call_order if item["call_id"] is not None} 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 c1f1586b0a6..3ae3bca0014 100644 --- a/python/packages/core/tests/core/test_function_invocation_logic.py +++ b/python/packages/core/tests/core/test_function_invocation_logic.py @@ -2286,6 +2286,68 @@ def func_rejected(arg1: str) -> str: assert msg.role == "tool", f"Message with FunctionResultContent must have role='tool', got '{msg.role}'" +async def test_stateless_sequential_approval_replay_preserves_model_order( + chat_client_base: SupportsChatGetResponse, +) -> None: + """Stateless approval replay should execute in request order, not response order.""" + execution_order: list[str] = [] + + @tool(name="first_write", approval_mode="always_require") + def first_write() -> str: + execution_order.append("first_write") + return "first" + + @tool(name="second_write", approval_mode="always_require") + def second_write() -> str: + execution_order.append("second_write") + return "second" + + chat_client_base.function_invocation_configuration["allow_concurrent_invocation"] = False # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + 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_first", name="first_write", arguments="{}"), + Content.from_function_call(call_id="call_second", name="second_write", arguments="{}"), + ], + ) + ), + ChatResponse(messages=Message(role="assistant", contents=["complete"])), + ] + options: ChatOptions[None] = {"tool_choice": "auto", "tools": [first_write, second_write]} + + first_response = await chat_client_base.get_response( + [Message(role="user", contents=["write in order"])], + options=options, + ) + approval_requests = [ + content + for message in first_response.messages + for content in message.contents + if content.type == "function_approval_request" + ] + assert [ + request.function_call.name for request in approval_requests if request.function_call is not None + ] == ["first_write", "second_write"] + + await chat_client_base.get_response( + [ + *first_response.messages, + Message( + role="user", + contents=[ + approval_requests[1].to_function_approval_response(approved=True), + approval_requests[0].to_function_approval_response(approved=True), + ], + ), + ], + options=options, + ) + + assert execution_order == ["first_write", "second_write"] + + 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 From afbf3cc07a9091baae17c604348883f258d04c6d Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 18 Sep 2026 08:50:50 +0200 Subject: [PATCH 07/10] fix: wait for complete approval batches --- .../specs/004-python-function-calling-loop.md | 4 +- .../packages/core/agent_framework/_tools.py | 115 +++++++++++++----- .../_workflows/_agent_executor.py | 6 +- .../tests/core/test_harness_tool_approval.py | 74 +++++++++++ 4 files changed, 165 insertions(+), 34 deletions(-) diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index ee3c44c6988..2b0c27f4fce 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -195,6 +195,8 @@ response is rejected, including when no response is supplied, message roles vary standalone request follows the incomplete batch. Stateless discovery selects the latest unresolved mixed batch, so a completed batch remains inert on later turns. Correlation is scoped to the active mixed batch so completed or abandoned historical Host calls remain unchanged. +Session-backed approval-only batches likewise remain atomic across resume calls: decisions may be accumulated, but no +call executes and the model is not resumed until every approval in the original batch has a decision. `ToolApprovalMiddleware` may resolve approval requests through standing or automatic policies, but it preserves non-approval user-input requests and does not split or reorder manual approvals relative to their Host-owned siblings. @@ -554,7 +556,7 @@ that manually replay messages own the equivalent rule: do not resend an approval | Reasoning-bound call | Finalized output retains reasoning, function call, function result, and final text. | `test_streaming_function_calling_response_includes_reasoning_and_tool_results` | | Calls across response messages | Every actionable call is executed once. | `test_base_client_executes_function_calls_across_multiple_response_messages` | | Parallel calls | Results retain the corresponding call ids and execution count. | `test_max_function_calls_limits_parallel_invocations`, `test_streaming_multiple_function_calls_parallel_execution` | -| Sequential calls | Calls execute one at a time in model order, including after approval replay; mixed local/Host-owned siblings retain that order, each call has isolated context, and middleware termination takes effect after the accepted batch finishes. | `packages/core/tests/core/test_tools.py::test_sequential_function_invocation_runs_calls_in_model_order`, `test_sequential_function_invocation_isolates_context`, `test_sequential_function_invocation_finishes_batch_after_termination`, `packages/core/tests/core/test_harness_tool_approval.py::test_mixed_batch_hides_already_approved_request_until_approval_replay`, `test_sequential_approval_replay_preserves_model_order_when_responses_are_reversed`, `packages/core/tests/core/test_function_invocation_logic.py::test_sequential_mixed_batch_preserves_approval_and_host_model_order` | +| Sequential calls | Calls execute one at a time in model order, including after approval replay; mixed local/Host-owned siblings retain that order, approval-only batches wait for every decision before executing, each call has isolated context, and middleware termination takes effect after the accepted batch finishes. | `packages/core/tests/core/test_tools.py::test_sequential_function_invocation_runs_calls_in_model_order`, `test_sequential_function_invocation_isolates_context`, `test_sequential_function_invocation_finishes_batch_after_termination`, `packages/core/tests/core/test_harness_tool_approval.py::test_mixed_batch_hides_already_approved_request_until_approval_replay`, `test_sequential_approval_replay_preserves_model_order_when_responses_are_reversed`, `test_sequential_approval_replay_waits_for_the_complete_batch`, `packages/core/tests/core/test_function_invocation_logic.py::test_sequential_mixed_batch_preserves_approval_and_host_model_order` | | Informational-only call | The call is returned but not executed or approved. | `test_informational_only_function_call_is_not_invoked`, `test_informational_only_function_call_does_not_request_approval`, `test_streaming_informational_only_function_call_is_not_invoked` | | OpenAI hosted/local shell boundary | Hosted shell calls remain informational in streaming and non-streaming responses even when a local executor is configured; only valid explicit local-shell items or shell calls marked with a local environment can execute, local execution preserves its configured approval mode, stateless loops preserve the complete provider shell transcript or fail explicitly, and locally generated shell outputs are sent in every continuation mode. | `packages/openai/tests/openai/test_openai_chat_client.py::test_response_content_creation_with_shell_call_remains_hosted_with_local_tool`, `test_parse_chunk_from_openai_shell_call_done_remains_hosted`, `test_parse_chunk_from_openai_local_environment_shell_call_done_emits_command`, `test_mixed_shell_calls_only_invoke_explicit_local_shell_call`, `test_stateless_shell_transcript_without_provider_item_fails`, `test_prepare_messages_keeps_local_shell_output_under_storage`, `test_malformed_local_environment_shell_call_is_not_executable`, `test_response_content_creation_with_local_shell_call_maps_to_function_call`, `test_malformed_local_shell_call_is_not_executable`, `test_response_function_call_named_local_shell_is_informational`, `test_parse_chunk_function_call_named_local_shell_is_informational`, `test_local_shell_tool_requires_approval_before_function_loop_execution` | | Declaration-only call | The call is surfaced as user input and is not executed; streaming arguments appear once while finalized request metadata remains available. | `test_declaration_only_tool`, `test_streaming_declaration_only_tool_preserves_metadata_without_duplicate_arguments` | diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index 2905456117e..c38d195d256 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -114,6 +114,7 @@ def _has_authoritative_approval_session(invocation_session: AgentSession | None) _ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY: Final[str] = "already_approved_approval_request_groups" +_APPROVAL_RESPONSES_KEY: Final[str] = "approval_responses" _FUNCTION_CALL_ORDER_KEY: Final[str] = "function_call_order" _PENDING_APPROVAL_REQUESTS_KEY: Final[str] = "pending_approval_requests" _PENDING_MIXED_PAUSE_BATCH_KEY: Final[str] = "pending_mixed_pause_batch" @@ -2916,33 +2917,76 @@ def _store_already_approved_approval_requests( state[_ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY] = pending_groups -def _pop_already_approved_approval_responses( +def _stage_approval_batch_responses( invocation_session: AgentSession | None, - approval_response_ids: set[str], -) -> tuple[list[Content], list[dict[str, str | None]]]: - """Pop the stored order and hidden safe requests for the approval ids being answered.""" - if not approval_response_ids: - return [], [] + approval_responses: Sequence[Content], +) -> tuple[list[Content], list[dict[str, str | None]], list[Content] | None]: + """Accumulate approval decisions and release a batch only when every decision is present.""" + if not approval_responses: + return [], [], None state = _get_tool_approval_state(invocation_session) if state is None: - return [], [] + return [], [], None raw_groups = state.get(_ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY, []) if not isinstance(raw_groups, list): - return [], [] + return [], [], None typed_groups = cast(list[Any], raw_groups) responses: list[Content] = [] function_call_order: list[dict[str, str | None]] = [] + waiting_requests: list[Content] | None = None remaining_groups: list[Any] = [] + pending_requests = _load_pending_approval_requests(invocation_session) for raw_group in typed_groups: if not isinstance(raw_group, Mapping): continue group = cast(Mapping[str, Any], 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): + group_ids = [str(item) for item in cast(list[Any], raw_ids)] if isinstance(raw_ids, list) else [] + if not group_ids: + continue + + stored_responses: dict[str, Content] = {} + raw_stored_responses = group.get(_APPROVAL_RESPONSES_KEY) + if isinstance(raw_stored_responses, list): + for raw_response in cast(list[Any], raw_stored_responses): + response = _content_from_state(raw_response) + if response is None: + continue + request_id = response.additional_properties.get(_APPROVAL_REQUEST_ID_KEY) + if isinstance(request_id, str) and request_id in group_ids: + stored_responses[request_id] = response + + matched = False + for response in approval_responses: + rebound = _bind_approval_response_to_pending_request(response, invocation_session, consume=False) + if rebound is None: + continue + request_id = rebound.additional_properties.get(_APPROVAL_REQUEST_ID_KEY) + if not isinstance(request_id, str) or request_id not in group_ids: + continue + stored_responses[request_id] = rebound + matched = True + + if not matched: remaining_groups.append(raw_group) continue + + if any(request_id not in stored_responses for request_id in group_ids): + updated_group = dict(group) + updated_group[_APPROVAL_RESPONSES_KEY] = [ + stored_responses[request_id].to_dict() + for request_id in group_ids + if request_id in stored_responses + ] + remaining_groups.append(updated_group) + missing_request_ids = [request_id for request_id in group_ids if request_id not in stored_responses] + if any(request_id not in pending_requests for request_id in missing_request_ids): + raise RuntimeError("An incomplete approval batch is missing one or more pending approval requests.") + waiting_requests = [pending_requests[request_id] for request_id in missing_request_ids] + continue + + responses.extend(stored_responses[request_id] for request_id in group_ids) raw_order = group.get(_FUNCTION_CALL_ORDER_KEY) if isinstance(raw_order, list): for raw_item in cast(list[Any], raw_order): @@ -2956,18 +3000,20 @@ def _pop_already_approved_approval_responses( "call_id": str(call_id) if call_id is not None else None, }) 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) - if request is None or request.type != "function_approval_request": - continue - responses.append(request.to_function_approval_response(approved=True)) + if isinstance(raw_requests, list): + for raw_request in cast(list[Any], raw_requests): + 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)) + for request_id in group_ids: + pending_requests.pop(request_id, None) if remaining_groups: state[_ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY] = remaining_groups else: state.pop(_ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY, None) - return responses, function_call_order + _save_pending_approval_requests(invocation_session, pending_requests) + return responses, function_call_order, waiting_requests def _store_pending_mixed_pause_batch( @@ -4116,31 +4162,36 @@ async def _resolve_approval_responses( "A mixed function-call batch requires responses for every approval and Host-owned request." ) + pending_responses_before_binding = list( + _collect_approval_responses(prepared_messages, non_approval_result_ids=host_result_ids).values() + ) + staged_responses, function_call_order, waiting_requests = _stage_approval_batch_responses( + approval_session, + pending_responses_before_binding, + ) + if waiting_requests is not None: + response_messages, streaming_updates = _messages_and_updates_for_terminal_contents(waiting_requests) + return _FunctionProcessingResult( + errors_in_a_row=errors_in_a_row, + action="return", + response_messages=response_messages, + streaming_updates=streaming_updates, + ) + + _bind_approval_responses_to_pending_requests(prepared_messages, approval_session) active_pending_ids = ( set(_load_pending_approval_requests(approval_session)) if _has_authoritative_approval_session(approval_session) else None ) - _bind_approval_responses_to_pending_requests(prepared_messages, approval_session) if completed_mixed_batch: state = _get_tool_approval_state(approval_session, create=False) if state is not None: state.pop(_PENDING_MIXED_PAUSE_BATCH_KEY, None) # 1. Restore safe siblings hidden with a prior mixed approval batch when its visible decision arrives. - explicit_approval_response_ids = { - content.id - for message in prepared_messages - for content in message.contents - if content.type == "function_approval_response" and content.id - } - - already_approved_responses, function_call_order = _pop_already_approved_approval_responses( - approval_session, - explicit_approval_response_ids, - ) - if already_approved_responses: - prepared_messages.append(Message(role="user", contents=already_approved_responses)) + if staged_responses: + prepared_messages.append(Message(role="user", contents=staged_responses)) if not function_call_order and approval_session is None and not host_result_ids: pending_stateless_responses = _collect_approval_responses(prepared_messages) function_call_order = _derive_stateless_approval_response_order( diff --git a/python/packages/core/agent_framework/_workflows/_agent_executor.py b/python/packages/core/agent_framework/_workflows/_agent_executor.py index 6867bd73a05..cd6dba7be2d 100644 --- a/python/packages/core/agent_framework/_workflows/_agent_executor.py +++ b/python/packages/core/agent_framework/_workflows/_agent_executor.py @@ -342,7 +342,11 @@ async def _cancel_pending_request( ctx: WorkflowContext[AgentExecutorResponse, AgentResponse | AgentResponseUpdate], ) -> None: """Release an agent-owned user-input request after workflow cancellation.""" - self._pending_agent_requests.pop(request_id, None) + cancelled_request = self._pending_agent_requests.pop(request_id, None) + if cancelled_request is not None and cancelled_request.type == "function_approval_request": + self._pending_responses_to_agent.append( + cancelled_request.to_function_approval_response(approved=False) + ) if not self._pending_agent_requests: await self._resume_with_pending_responses(ctx) diff --git a/python/packages/core/tests/core/test_harness_tool_approval.py b/python/packages/core/tests/core/test_harness_tool_approval.py index 4e5dd99f1c5..e5c38da75d1 100644 --- a/python/packages/core/tests/core/test_harness_tool_approval.py +++ b/python/packages/core/tests/core/test_harness_tool_approval.py @@ -1608,6 +1608,80 @@ def second_write() -> str: assert execution_order == ["first_write", "second_write"] +@pytest.mark.parametrize( + ("first_approved", "expected_execution_order"), + [ + (True, ["first_write", "second_read"]), + (False, ["second_read"]), + ], +) +async def test_sequential_approval_replay_waits_for_the_complete_batch( + chat_client_base: MockBaseChatClient, + first_approved: bool, + expected_execution_order: list[str], +) -> None: + """A partial approval batch must not execute tools or advance the model.""" + execution_order: list[str] = [] + + @tool(name="first_write", approval_mode="always_require") + def first_write() -> str: + execution_order.append("first_write") + return "first" + + @tool(name="second_read", approval_mode="always_require") + def second_read() -> str: + execution_order.append("second_read") + return "second" + + agent = Agent(client=chat_client_base, tools=[first_write, second_read]) + chat_client_base.function_invocation_configuration["allow_concurrent_invocation"] = False + session = AgentSession(session_id=f"partial-approval-order-{first_approved}") + chat_client_base.run_responses = [ + ChatResponse( + messages=Message( + role="assistant", + contents=[ + Content.from_function_call(call_id="call_first", name="first_write", arguments="{}"), + Content.from_function_call(call_id="call_second", name="second_read", arguments="{}"), + ], + ) + ) + ] + + first_response = await agent.run("write then read", session=session) + requests = _approval_requests(first_response.messages) + assert [_function_call(request).name for request in requests] == ["first_write", "second_read"] + + partial_response = await agent.run( + requests[1].to_function_approval_response(approved=True), + session=session, + ) + + assert execution_order == [] + assert [ + _function_call(request).name for request in _approval_requests(partial_response.messages) + ] == ["first_write"] + + repeated_partial_response = await agent.run( + requests[1].to_function_approval_response(approved=True), + session=session, + ) + + assert execution_order == [] + assert [_function_call(request).name for request in _approval_requests(repeated_partial_response.messages)] == [ + "first_write" + ] + + chat_client_base.run_responses = [ChatResponse(messages=Message(role="assistant", contents=["complete"]))] + final_response = await agent.run( + requests[0].to_function_approval_response(approved=first_approved), + session=session, + ) + + assert final_response.text == "complete" + assert execution_order == expected_execution_order + + async def test_mixed_batch_accepts_restored_tool_approval_state( chat_client_base: MockBaseChatClient, ) -> None: From 36c3bd8c84b34757dff0a7524733d6ada9898c60 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 18 Sep 2026 09:11:42 +0200 Subject: [PATCH 08/10] fix: settle duplicate and cancelled batch responses --- .../specs/004-python-function-calling-loop.md | 4 +- .../packages/core/agent_framework/_tools.py | 11 +-- .../_workflows/_agent_executor.py | 12 +++ .../tests/core/test_harness_tool_approval.py | 51 +++++++++++++ .../test_agent_executor_tool_calls.py | 76 ++++++++++++++++++- 5 files changed, 144 insertions(+), 10 deletions(-) diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index 2b0c27f4fce..b3a7715c385 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -556,7 +556,7 @@ that manually replay messages own the equivalent rule: do not resend an approval | Reasoning-bound call | Finalized output retains reasoning, function call, function result, and final text. | `test_streaming_function_calling_response_includes_reasoning_and_tool_results` | | Calls across response messages | Every actionable call is executed once. | `test_base_client_executes_function_calls_across_multiple_response_messages` | | Parallel calls | Results retain the corresponding call ids and execution count. | `test_max_function_calls_limits_parallel_invocations`, `test_streaming_multiple_function_calls_parallel_execution` | -| Sequential calls | Calls execute one at a time in model order, including after approval replay; mixed local/Host-owned siblings retain that order, approval-only batches wait for every decision before executing, each call has isolated context, and middleware termination takes effect after the accepted batch finishes. | `packages/core/tests/core/test_tools.py::test_sequential_function_invocation_runs_calls_in_model_order`, `test_sequential_function_invocation_isolates_context`, `test_sequential_function_invocation_finishes_batch_after_termination`, `packages/core/tests/core/test_harness_tool_approval.py::test_mixed_batch_hides_already_approved_request_until_approval_replay`, `test_sequential_approval_replay_preserves_model_order_when_responses_are_reversed`, `test_sequential_approval_replay_waits_for_the_complete_batch`, `packages/core/tests/core/test_function_invocation_logic.py::test_sequential_mixed_batch_preserves_approval_and_host_model_order` | +| Sequential calls | Calls execute one at a time in model order, including after approval replay; mixed local/Host-owned siblings retain that order, approval-only batches wait for every decision before executing, each call has isolated context, and middleware termination takes effect after the accepted batch finishes. | `packages/core/tests/core/test_tools.py::test_sequential_function_invocation_runs_calls_in_model_order`, `test_sequential_function_invocation_isolates_context`, `test_sequential_function_invocation_finishes_batch_after_termination`, `packages/core/tests/core/test_harness_tool_approval.py::test_mixed_batch_hides_already_approved_request_until_approval_replay`, `test_sequential_approval_replay_preserves_model_order_when_responses_are_reversed`, `test_sequential_approval_replay_waits_for_the_complete_batch`, `packages/core/tests/core/test_function_invocation_logic.py::test_sequential_mixed_batch_preserves_approval_and_host_model_order`, `packages/core/tests/workflow/test_agent_executor_tool_calls.py::test_workflow_cancels_host_member_of_mixed_batch_with_terminal_result` | | Informational-only call | The call is returned but not executed or approved. | `test_informational_only_function_call_is_not_invoked`, `test_informational_only_function_call_does_not_request_approval`, `test_streaming_informational_only_function_call_is_not_invoked` | | OpenAI hosted/local shell boundary | Hosted shell calls remain informational in streaming and non-streaming responses even when a local executor is configured; only valid explicit local-shell items or shell calls marked with a local environment can execute, local execution preserves its configured approval mode, stateless loops preserve the complete provider shell transcript or fail explicitly, and locally generated shell outputs are sent in every continuation mode. | `packages/openai/tests/openai/test_openai_chat_client.py::test_response_content_creation_with_shell_call_remains_hosted_with_local_tool`, `test_parse_chunk_from_openai_shell_call_done_remains_hosted`, `test_parse_chunk_from_openai_local_environment_shell_call_done_emits_command`, `test_mixed_shell_calls_only_invoke_explicit_local_shell_call`, `test_stateless_shell_transcript_without_provider_item_fails`, `test_prepare_messages_keeps_local_shell_output_under_storage`, `test_malformed_local_environment_shell_call_is_not_executable`, `test_response_content_creation_with_local_shell_call_maps_to_function_call`, `test_malformed_local_shell_call_is_not_executable`, `test_response_function_call_named_local_shell_is_informational`, `test_parse_chunk_function_call_named_local_shell_is_informational`, `test_local_shell_tool_requires_approval_before_function_loop_execution` | | Declaration-only call | The call is surfaced as user input and is not executed; streaming arguments appear once while finalized request metadata remains available. | `test_declaration_only_tool`, `test_streaming_declaration_only_tool_preserves_metadata_without_duplicate_arguments` | @@ -585,7 +585,7 @@ that manually replay messages own the equivalent rule: do not resend an approval | Changed resolved policy invocation | The stale decision executes nothing; a same-occurrence replacement request is visible and persisted in both modes, and the second approval executes exactly once. | `packages/core/tests/core/test_harness_tool_approval.py::test_changed_hidden_snapshot_requires_visible_second_approval` | | Expired or evicted policy authority | The old response executes nothing, surfaces and persists a same-occurrence replacement with a fresh request generation in both modes, and executes exactly once only after the replacement is approved; restored model history remains balanced and stale prior-generation replay is inert. | `packages/core/tests/core/test_harness_tool_approval.py::test_policy_reapproval_is_visible_persisted_and_executes_once`, `packages/core/tests/test_security.py::TestPolicyEnforcementMiddleware::test_pending_policy_approval_ttl_is_deterministic_and_durable` | | Session-bound policy cleanup | FIFO/TTL lifecycle and authenticated rejection/cancellation cleanup use occurrence identity within only the owning session, including fixed provider scopes. | `packages/core/tests/test_security.py::TestPolicyEnforcementMiddleware::test_pending_policy_approvals_are_fifo_bounded_by_occurrence`, `test_pending_policy_approval_ttl_is_deterministic_and_durable`, `test_non_grant_cleanup_is_authenticated_session_and_occurrence_bound`, `test_fixed_scope_non_grant_cleanup_keeps_unrelated_occurrence` | -| Session-bound substituted response | A response is rebound to the immutable recorded call and cannot replace its call id, tool name, or arguments. | `packages/core/tests/core/test_function_invocation_logic.py::test_session_approval_binding_rebinds_consumes_and_rejects_duplicates` | +| Session-bound substituted response | A response is rebound to the immutable recorded call and cannot replace its call id, tool name, arguments, or first decision through a conflicting duplicate. | `packages/core/tests/core/test_function_invocation_logic.py::test_session_approval_binding_rebinds_consumes_and_rejects_duplicates`, `packages/core/tests/core/test_harness_tool_approval.py::test_approval_batch_uses_the_first_decision_for_duplicate_responses` | | Occurrence-aware local binding | New local requests use `function_call.id`; missing, mismatched, or stale occurrence ids do not execute or consume pending state, while the canonical occurrence id binds without an embedded call. | `test_occurrence_aware_approval_rejects_stale_reused_call_id_response`, `test_occurrence_aware_approval_mismatched_identity_does_not_consume_pending`, `test_occurrence_aware_approval_binds_without_embedded_function_call` | | Legacy stored approval | A serialized pending request without `function_call.id` retains exact request-id binding once and warns only when resumed. | `test_legacy_serialized_pending_approval_resumes_once_with_migration_warning`, `packages/core/tests/core/test_types.py::test_legacy_function_call_deserialization_does_not_generate_an_occurrence_id` | | Hosted approval identity | Provider-issued hosted approval request ids are unchanged by local occurrence correlation. | `test_hosted_approval_keeps_provider_issued_request_id` | diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index c38d195d256..61ad835de84 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -3361,11 +3361,12 @@ def _collect_approval_responses( if resolved.id is not None and pending_by_approval_id.get(resolved.id) is resolved: pending_by_approval_id.pop(resolved.id, None) - return { - content.id: content - for content in approval_responses - if id(content) not in resolved_response_ids and content.id is not None - } + collected_responses: dict[str, Content] = {} + for content in approval_responses: + if id(content) in resolved_response_ids or content.id is None: + continue + collected_responses.setdefault(content.id, content) + return collected_responses def _collect_unanswered_approval_requests(messages: Sequence[Message]) -> list[Content]: diff --git a/python/packages/core/agent_framework/_workflows/_agent_executor.py b/python/packages/core/agent_framework/_workflows/_agent_executor.py index cd6dba7be2d..d92f77aa699 100644 --- a/python/packages/core/agent_framework/_workflows/_agent_executor.py +++ b/python/packages/core/agent_framework/_workflows/_agent_executor.py @@ -347,6 +347,18 @@ async def _cancel_pending_request( self._pending_responses_to_agent.append( cancelled_request.to_function_approval_response(approved=False) ) + elif ( + cancelled_request is not None + and cancelled_request.type == "function_call" + and cancelled_request.call_id is not None + ): + self._pending_responses_to_agent.append( + Content.from_function_result( + call_id=cancelled_request.call_id, + result="Error: Tool call was cancelled.", + additional_properties={"cancelled": True}, + ) + ) if not self._pending_agent_requests: await self._resume_with_pending_responses(ctx) diff --git a/python/packages/core/tests/core/test_harness_tool_approval.py b/python/packages/core/tests/core/test_harness_tool_approval.py index e5c38da75d1..43040e318da 100644 --- a/python/packages/core/tests/core/test_harness_tool_approval.py +++ b/python/packages/core/tests/core/test_harness_tool_approval.py @@ -1682,6 +1682,57 @@ def second_read() -> str: assert execution_order == expected_execution_order +async def test_approval_batch_uses_the_first_decision_for_duplicate_responses( + chat_client_base: MockBaseChatClient, +) -> None: + """A conflicting duplicate must not replace the first decision for an approval.""" + execution_order: list[str] = [] + + @tool(name="first_write", approval_mode="always_require") + def first_write() -> str: + execution_order.append("first_write") + return "first" + + @tool(name="second_write", approval_mode="always_require") + def second_write() -> str: + execution_order.append("second_write") + return "second" + + agent = Agent(client=chat_client_base, tools=[first_write, second_write]) + session = AgentSession(session_id="duplicate-approval-decisions") + chat_client_base.run_responses = [ + ChatResponse( + messages=Message( + role="assistant", + contents=[ + Content.from_function_call(call_id="call_first", name="first_write", arguments="{}"), + Content.from_function_call(call_id="call_second", name="second_write", arguments="{}"), + ], + ) + ) + ] + + first_response = await agent.run("write twice", session=session) + requests = _approval_requests(first_response.messages) + assert [_function_call(request).name for request in requests] == ["first_write", "second_write"] + + chat_client_base.run_responses = [ChatResponse(messages=Message(role="assistant", contents=["complete"]))] + final_response = await agent.run( + Message( + role="user", + contents=[ + requests[0].to_function_approval_response(approved=False), + requests[0].to_function_approval_response(approved=True), + requests[1].to_function_approval_response(approved=True), + ], + ), + session=session, + ) + + assert final_response.text == "complete" + assert execution_order == ["second_write"] + + async def test_mixed_batch_accepts_restored_tool_approval_state( chat_client_base: MockBaseChatClient, ) -> None: diff --git a/python/packages/core/tests/workflow/test_agent_executor_tool_calls.py b/python/packages/core/tests/workflow/test_agent_executor_tool_calls.py index cee0b1570e8..a3ea70d9599 100644 --- a/python/packages/core/tests/workflow/test_agent_executor_tool_calls.py +++ b/python/packages/core/tests/workflow/test_agent_executor_tool_calls.py @@ -511,11 +511,12 @@ async def test_agent_executor_parallel_tool_call_with_approval_streaming() -> No class DeclarationOnlyMockChatClient(FunctionInvocationLayer[Any], BaseChatClient[Any]): """Mock chat client that calls a declaration-only tool on first iteration.""" - def __init__(self, parallel_request: bool = False) -> None: + def __init__(self, parallel_request: bool = False, mixed_request: bool = False) -> None: FunctionInvocationLayer.__init__(self) BaseChatClient.__init__(self) self._iteration: int = 0 self._parallel_request: bool = parallel_request + self._mixed_request: bool = mixed_request self.received_messages: list[list[Message]] = [] def _inner_get_response( @@ -537,7 +538,25 @@ async def _get_response() -> ChatResponse: def _create_response(self) -> ChatResponse: if self._iteration == 0: - if self._parallel_request: + if self._mixed_request: + response = ChatResponse( + messages=Message( + "assistant", + [ + Content.from_function_call( + call_id="approval-call", + name="mock_tool_requiring_approval", + arguments='{"query": "approved"}', + ), + Content.from_function_call( + call_id="host-call", + name="client_side_tool", + arguments='{"query": "hosted"}', + ), + ], + ) + ) + elif self._parallel_request: response = ChatResponse( messages=Message( "assistant", @@ -570,7 +589,23 @@ def _create_response(self) -> ChatResponse: async def _stream_response(self) -> AsyncIterable[ChatResponseUpdate]: if self._iteration == 0: - if self._parallel_request: + if self._mixed_request: + yield ChatResponseUpdate( + contents=[ + Content.from_function_call( + call_id="approval-call", + name="mock_tool_requiring_approval", + arguments='{"query": "approved"}', + ), + Content.from_function_call( + call_id="host-call", + name="client_side_tool", + arguments='{"query": "hosted"}', + ), + ], + role="assistant", + ) + elif self._parallel_request: yield ChatResponseUpdate( contents=[ Content.from_function_call(call_id="1", name="client_side_tool", arguments='{"query": "test"}'), @@ -728,3 +763,38 @@ async def test_agent_executor_parallel_declaration_only_tool_emits_request_info( final_response = events.get_outputs() assert len(final_response) == 1 assert final_response[0] == "Tool executed successfully." + + +async def test_workflow_cancels_host_member_of_mixed_batch_with_terminal_result() -> None: + """Cancelling a Host-owned sibling completes the mixed batch with its exact call identity.""" + client = DeclarationOnlyMockChatClient(mixed_request=True) + agent = Agent( + client=client, + name="MixedApprovalAgent", + tools=[mock_tool_requiring_approval, declaration_only_tool], + ) + workflow = WorkflowBuilder(start_executor=agent, output_from=[test_executor]).add_edge(agent, test_executor).build() + + paused = await workflow.run("Run both tools") + requests = paused.get_request_info_events() + approval_request = next(request for request in requests if request.data.type == "function_approval_request") + host_request = next(request for request in requests if request.data.type == "function_call") + + cancelled = await workflow.cancel_pending_requests([host_request.request_id]) + assert cancelled.get_outputs() == [] + + resumed = await workflow.run( + responses={ + approval_request.request_id: approval_request.data.to_function_approval_response(approved=True), + } + ) + + assert resumed.get_outputs() == ["Tool executed successfully."] + function_results = [ + content + for message in client.received_messages[-1] + for content in message.contents + if content.type == "function_result" + ] + cancelled_result = next(result for result in function_results if result.call_id == host_request.data.call_id) + assert cancelled_result.additional_properties["cancelled"] is True From 1e40a90419f16501808d2bc8eec6018f457df619 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 18 Sep 2026 09:25:42 +0200 Subject: [PATCH 09/10] fix: preserve staged approval decisions --- .../specs/004-python-function-calling-loop.md | 2 +- .../packages/core/agent_framework/_tools.py | 3 +- .../tests/core/test_harness_tool_approval.py | 58 +++++++++++++++++++ 3 files changed, 61 insertions(+), 2 deletions(-) diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index b3a7715c385..ecc3495f9ed 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -585,7 +585,7 @@ that manually replay messages own the equivalent rule: do not resend an approval | Changed resolved policy invocation | The stale decision executes nothing; a same-occurrence replacement request is visible and persisted in both modes, and the second approval executes exactly once. | `packages/core/tests/core/test_harness_tool_approval.py::test_changed_hidden_snapshot_requires_visible_second_approval` | | Expired or evicted policy authority | The old response executes nothing, surfaces and persists a same-occurrence replacement with a fresh request generation in both modes, and executes exactly once only after the replacement is approved; restored model history remains balanced and stale prior-generation replay is inert. | `packages/core/tests/core/test_harness_tool_approval.py::test_policy_reapproval_is_visible_persisted_and_executes_once`, `packages/core/tests/test_security.py::TestPolicyEnforcementMiddleware::test_pending_policy_approval_ttl_is_deterministic_and_durable` | | Session-bound policy cleanup | FIFO/TTL lifecycle and authenticated rejection/cancellation cleanup use occurrence identity within only the owning session, including fixed provider scopes. | `packages/core/tests/test_security.py::TestPolicyEnforcementMiddleware::test_pending_policy_approvals_are_fifo_bounded_by_occurrence`, `test_pending_policy_approval_ttl_is_deterministic_and_durable`, `test_non_grant_cleanup_is_authenticated_session_and_occurrence_bound`, `test_fixed_scope_non_grant_cleanup_keeps_unrelated_occurrence` | -| Session-bound substituted response | A response is rebound to the immutable recorded call and cannot replace its call id, tool name, arguments, or first decision through a conflicting duplicate. | `packages/core/tests/core/test_function_invocation_logic.py::test_session_approval_binding_rebinds_consumes_and_rejects_duplicates`, `packages/core/tests/core/test_harness_tool_approval.py::test_approval_batch_uses_the_first_decision_for_duplicate_responses` | +| Session-bound substituted response | A response is rebound to the immutable recorded call and cannot replace its call id, tool name, arguments, or first decision through a conflicting duplicate in the same request or a later retry. | `packages/core/tests/core/test_function_invocation_logic.py::test_session_approval_binding_rebinds_consumes_and_rejects_duplicates`, `packages/core/tests/core/test_harness_tool_approval.py::test_approval_batch_uses_the_first_decision_for_duplicate_responses`, `test_approval_batch_preserves_first_decision_across_partial_retries` | | Occurrence-aware local binding | New local requests use `function_call.id`; missing, mismatched, or stale occurrence ids do not execute or consume pending state, while the canonical occurrence id binds without an embedded call. | `test_occurrence_aware_approval_rejects_stale_reused_call_id_response`, `test_occurrence_aware_approval_mismatched_identity_does_not_consume_pending`, `test_occurrence_aware_approval_binds_without_embedded_function_call` | | Legacy stored approval | A serialized pending request without `function_call.id` retains exact request-id binding once and warns only when resumed. | `test_legacy_serialized_pending_approval_resumes_once_with_migration_warning`, `packages/core/tests/core/test_types.py::test_legacy_function_call_deserialization_does_not_generate_an_occurrence_id` | | Hosted approval identity | Provider-issued hosted approval request ids are unchanged by local occurrence correlation. | `test_hosted_approval_keeps_provider_issued_request_id` | diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index 61ad835de84..a9f878c05af 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -2965,7 +2965,8 @@ def _stage_approval_batch_responses( request_id = rebound.additional_properties.get(_APPROVAL_REQUEST_ID_KEY) if not isinstance(request_id, str) or request_id not in group_ids: continue - stored_responses[request_id] = rebound + if request_id not in stored_responses: + stored_responses[request_id] = rebound matched = True if not matched: diff --git a/python/packages/core/tests/core/test_harness_tool_approval.py b/python/packages/core/tests/core/test_harness_tool_approval.py index 43040e318da..cf5e179153c 100644 --- a/python/packages/core/tests/core/test_harness_tool_approval.py +++ b/python/packages/core/tests/core/test_harness_tool_approval.py @@ -1733,6 +1733,64 @@ def second_write() -> str: assert execution_order == ["second_write"] +async def test_approval_batch_preserves_first_decision_across_partial_retries( + chat_client_base: MockBaseChatClient, +) -> None: + """A later retry must not replace a decision staged by an earlier partial response.""" + execution_order: list[str] = [] + + @tool(name="first_write", approval_mode="always_require") + def first_write() -> str: + execution_order.append("first_write") + return "first" + + @tool(name="second_write", approval_mode="always_require") + def second_write() -> str: + execution_order.append("second_write") + return "second" + + agent = Agent(client=chat_client_base, tools=[first_write, second_write]) + session = AgentSession(session_id="partial-retry-conflicting-decision") + chat_client_base.run_responses = [ + ChatResponse( + messages=Message( + role="assistant", + contents=[ + Content.from_function_call(call_id="call_first", name="first_write", arguments="{}"), + Content.from_function_call(call_id="call_second", name="second_write", arguments="{}"), + ], + ) + ) + ] + + first_response = await agent.run("write twice", session=session) + requests = _approval_requests(first_response.messages) + partial_response = await agent.run( + requests[1].to_function_approval_response(approved=False), + session=session, + ) + + assert execution_order == [] + assert [_function_call(request).name for request in _approval_requests(partial_response.messages)] == [ + "first_write" + ] + + chat_client_base.run_responses = [ChatResponse(messages=Message(role="assistant", contents=["complete"]))] + final_response = await agent.run( + Message( + role="user", + contents=[ + requests[0].to_function_approval_response(approved=True), + requests[1].to_function_approval_response(approved=True), + ], + ), + session=session, + ) + + assert final_response.text == "complete" + assert execution_order == ["first_write"] + + async def test_mixed_batch_accepts_restored_tool_approval_state( chat_client_base: MockBaseChatClient, ) -> None: From b376b862e865d025dba7fa59a93d75b79f053892 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 18 Sep 2026 09:42:27 +0200 Subject: [PATCH 10/10] fix: preserve cancelled Host occurrence identity --- .../specs/004-python-function-calling-loop.md | 2 +- .../_workflows/_agent_executor.py | 12 +- .../test_agent_executor_tool_calls.py | 108 +++++++++++++----- 3 files changed, 87 insertions(+), 35 deletions(-) diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index ecc3495f9ed..370e6c11376 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -556,7 +556,7 @@ that manually replay messages own the equivalent rule: do not resend an approval | Reasoning-bound call | Finalized output retains reasoning, function call, function result, and final text. | `test_streaming_function_calling_response_includes_reasoning_and_tool_results` | | Calls across response messages | Every actionable call is executed once. | `test_base_client_executes_function_calls_across_multiple_response_messages` | | Parallel calls | Results retain the corresponding call ids and execution count. | `test_max_function_calls_limits_parallel_invocations`, `test_streaming_multiple_function_calls_parallel_execution` | -| Sequential calls | Calls execute one at a time in model order, including after approval replay; mixed local/Host-owned siblings retain that order, approval-only batches wait for every decision before executing, each call has isolated context, and middleware termination takes effect after the accepted batch finishes. | `packages/core/tests/core/test_tools.py::test_sequential_function_invocation_runs_calls_in_model_order`, `test_sequential_function_invocation_isolates_context`, `test_sequential_function_invocation_finishes_batch_after_termination`, `packages/core/tests/core/test_harness_tool_approval.py::test_mixed_batch_hides_already_approved_request_until_approval_replay`, `test_sequential_approval_replay_preserves_model_order_when_responses_are_reversed`, `test_sequential_approval_replay_waits_for_the_complete_batch`, `packages/core/tests/core/test_function_invocation_logic.py::test_sequential_mixed_batch_preserves_approval_and_host_model_order`, `packages/core/tests/workflow/test_agent_executor_tool_calls.py::test_workflow_cancels_host_member_of_mixed_batch_with_terminal_result` | +| Sequential calls | Calls execute one at a time in model order, including after approval replay; mixed local/Host-owned siblings retain that order, approval-only batches wait for every decision before executing, each call has isolated context, and middleware termination takes effect after the accepted batch finishes. | `packages/core/tests/core/test_tools.py::test_sequential_function_invocation_runs_calls_in_model_order`, `test_sequential_function_invocation_isolates_context`, `test_sequential_function_invocation_finishes_batch_after_termination`, `packages/core/tests/core/test_harness_tool_approval.py::test_mixed_batch_hides_already_approved_request_until_approval_replay`, `test_sequential_approval_replay_preserves_model_order_when_responses_are_reversed`, `test_sequential_approval_replay_waits_for_the_complete_batch`, `packages/core/tests/core/test_function_invocation_logic.py::test_sequential_mixed_batch_preserves_approval_and_host_model_order`, `packages/core/tests/workflow/test_agent_executor_tool_calls.py::test_workflow_cancels_host_member_of_mixed_batch_with_terminal_result`, `test_workflow_host_cancellation_preserves_occurrence_with_reused_call_id` | | Informational-only call | The call is returned but not executed or approved. | `test_informational_only_function_call_is_not_invoked`, `test_informational_only_function_call_does_not_request_approval`, `test_streaming_informational_only_function_call_is_not_invoked` | | OpenAI hosted/local shell boundary | Hosted shell calls remain informational in streaming and non-streaming responses even when a local executor is configured; only valid explicit local-shell items or shell calls marked with a local environment can execute, local execution preserves its configured approval mode, stateless loops preserve the complete provider shell transcript or fail explicitly, and locally generated shell outputs are sent in every continuation mode. | `packages/openai/tests/openai/test_openai_chat_client.py::test_response_content_creation_with_shell_call_remains_hosted_with_local_tool`, `test_parse_chunk_from_openai_shell_call_done_remains_hosted`, `test_parse_chunk_from_openai_local_environment_shell_call_done_emits_command`, `test_mixed_shell_calls_only_invoke_explicit_local_shell_call`, `test_stateless_shell_transcript_without_provider_item_fails`, `test_prepare_messages_keeps_local_shell_output_under_storage`, `test_malformed_local_environment_shell_call_is_not_executable`, `test_response_content_creation_with_local_shell_call_maps_to_function_call`, `test_malformed_local_shell_call_is_not_executable`, `test_response_function_call_named_local_shell_is_informational`, `test_parse_chunk_function_call_named_local_shell_is_informational`, `test_local_shell_tool_requires_approval_before_function_loop_execution` | | Declaration-only call | The call is surfaced as user input and is not executed; streaming arguments appear once while finalized request metadata remains available. | `test_declaration_only_tool`, `test_streaming_declaration_only_tool_preserves_metadata_without_duplicate_arguments` | diff --git a/python/packages/core/agent_framework/_workflows/_agent_executor.py b/python/packages/core/agent_framework/_workflows/_agent_executor.py index d92f77aa699..a56cf2d0331 100644 --- a/python/packages/core/agent_framework/_workflows/_agent_executor.py +++ b/python/packages/core/agent_framework/_workflows/_agent_executor.py @@ -352,13 +352,13 @@ async def _cancel_pending_request( and cancelled_request.type == "function_call" and cancelled_request.call_id is not None ): - self._pending_responses_to_agent.append( - Content.from_function_result( - call_id=cancelled_request.call_id, - result="Error: Tool call was cancelled.", - additional_properties={"cancelled": True}, - ) + cancellation_result = Content.from_function_result( + call_id=cancelled_request.call_id, + result="Error: Tool call was cancelled.", + additional_properties={"cancelled": True}, ) + cancellation_result.id = cancelled_request.id + self._pending_responses_to_agent.append(cancellation_result) if not self._pending_agent_requests: await self._resume_with_pending_responses(ctx) diff --git a/python/packages/core/tests/workflow/test_agent_executor_tool_calls.py b/python/packages/core/tests/workflow/test_agent_executor_tool_calls.py index a3ea70d9599..46278205945 100644 --- a/python/packages/core/tests/workflow/test_agent_executor_tool_calls.py +++ b/python/packages/core/tests/workflow/test_agent_executor_tool_calls.py @@ -511,14 +511,43 @@ async def test_agent_executor_parallel_tool_call_with_approval_streaming() -> No class DeclarationOnlyMockChatClient(FunctionInvocationLayer[Any], BaseChatClient[Any]): """Mock chat client that calls a declaration-only tool on first iteration.""" - def __init__(self, parallel_request: bool = False, mixed_request: bool = False) -> None: + def __init__( + self, + parallel_request: bool = False, + mixed_request: bool = False, + duplicate_host_call_id: bool = False, + ) -> None: FunctionInvocationLayer.__init__(self) BaseChatClient.__init__(self) self._iteration: int = 0 self._parallel_request: bool = parallel_request self._mixed_request: bool = mixed_request + self._duplicate_host_call_id: bool = duplicate_host_call_id self.received_messages: list[list[Message]] = [] + def _mixed_request_contents(self) -> list[Content]: + contents = [ + Content.from_function_call( + call_id="approval-call", + name="mock_tool_requiring_approval", + arguments='{"query": "approved"}', + ), + Content.from_function_call( + call_id="host-call", + name="client_side_tool", + arguments='{"query": "hosted"}', + ), + ] + if self._duplicate_host_call_id: + contents.append( + Content.from_function_call( + call_id="host-call", + name="client_side_tool", + arguments='{"query": "also-hosted"}', + ) + ) + return contents + def _inner_get_response( self, *, @@ -540,21 +569,7 @@ def _create_response(self) -> ChatResponse: if self._iteration == 0: if self._mixed_request: response = ChatResponse( - messages=Message( - "assistant", - [ - Content.from_function_call( - call_id="approval-call", - name="mock_tool_requiring_approval", - arguments='{"query": "approved"}', - ), - Content.from_function_call( - call_id="host-call", - name="client_side_tool", - arguments='{"query": "hosted"}', - ), - ], - ) + messages=Message("assistant", self._mixed_request_contents()) ) elif self._parallel_request: response = ChatResponse( @@ -591,18 +606,7 @@ async def _stream_response(self) -> AsyncIterable[ChatResponseUpdate]: if self._iteration == 0: if self._mixed_request: yield ChatResponseUpdate( - contents=[ - Content.from_function_call( - call_id="approval-call", - name="mock_tool_requiring_approval", - arguments='{"query": "approved"}', - ), - Content.from_function_call( - call_id="host-call", - name="client_side_tool", - arguments='{"query": "hosted"}', - ), - ], + contents=self._mixed_request_contents(), role="assistant", ) elif self._parallel_request: @@ -798,3 +802,51 @@ async def test_workflow_cancels_host_member_of_mixed_batch_with_terminal_result( ] cancelled_result = next(result for result in function_results if result.call_id == host_request.data.call_id) assert cancelled_result.additional_properties["cancelled"] is True + + +async def test_workflow_host_cancellation_preserves_occurrence_with_reused_call_id() -> None: + """A Host cancellation identifies the exact occurrence when sibling call IDs are reused.""" + client = DeclarationOnlyMockChatClient(mixed_request=True, duplicate_host_call_id=True) + agent = Agent( + client=client, + name="MixedApprovalAgent", + tools=[mock_tool_requiring_approval, declaration_only_tool], + ) + workflow = WorkflowBuilder(start_executor=agent, output_from=[test_executor]).add_edge(agent, test_executor).build() + + paused = await workflow.run("Run all tools") + requests = paused.get_request_info_events() + approval_request = next(request for request in requests if request.data.type == "function_approval_request") + host_requests = [request for request in requests if request.data.type == "function_call"] + assert len(host_requests) == 2 + assert host_requests[0].data.call_id == host_requests[1].data.call_id + assert host_requests[0].data.id != host_requests[1].data.id + + cancelled = await workflow.cancel_pending_requests([host_requests[0].request_id]) + assert cancelled.get_outputs() == [] + + sibling_result = Content.from_function_result( + call_id=host_requests[1].data.call_id, + result="Host sibling completed.", + ) + sibling_result.id = host_requests[1].data.id + resumed = await workflow.run( + responses={ + approval_request.request_id: approval_request.data.to_function_approval_response(approved=True), + host_requests[1].request_id: sibling_result, + } + ) + + assert resumed.get_outputs() == ["Tool executed successfully."] + function_results = [ + content + for message in client.received_messages[-1] + for content in message.contents + if content.type == "function_result" and content.call_id == host_requests[0].data.call_id + ] + assert {result.id for result in function_results} == { + host_requests[0].data.id, + host_requests[1].data.id, + } + cancelled_result = next(result for result in function_results if result.id == host_requests[0].data.id) + assert cancelled_result.additional_properties["cancelled"] is True