diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md
index 4603417996c..370e6c11376 100644
--- a/docs/specs/004-python-function-calling-loop.md
+++ b/docs/specs/004-python-function-calling-loop.md
@@ -195,9 +195,16 @@ 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.
 
+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 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
 
 Some hosted services bind reasoning content or an opaque reasoning signature to the function call that follows it.
@@ -549,6 +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`, `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` |
@@ -577,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 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 a292383ac7a..a9f878c05af 100644
--- a/python/packages/core/agent_framework/_tools.py
+++ b/python/packages/core/agent_framework/_tools.py
@@ -114,6 +114,8 @@ 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"
 _APPROVAL_REQUEST_ID_KEY: Final[str] = "_approval_request_id"
@@ -1693,6 +1695,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
@@ -1732,6 +1737,7 @@ class FunctionInvocationConfiguration(TypedDict, total=False):
     terminate_on_unknown_calls: bool
     additional_tools: Sequence[FunctionTool]
     include_detailed_errors: bool
+    allow_concurrent_invocation: bool
 
 
 def normalize_function_invocation_configuration(
@@ -1746,6 +1752,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)
@@ -2339,6 +2346,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] = []
+        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":
@@ -2373,6 +2385,7 @@ async def _try_execute_function_call_groups(
             approval_session,
             visible_requests,
             already_approved_requests,
+            function_call_order=function_call_order,
         )
         _store_pending_approval_requests(approval_session, visible_requests)
         _store_pending_mixed_pause_batch(approval_session, pause_groups)
@@ -2387,11 +2400,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,
@@ -2404,22 +2416,28 @@ 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 function_call in function_calls:
+            result = await create_execution_task(function_call)
+            execution_results.append(result)
 
     should_terminate = any(terminate for _, terminate in execution_results)
     return [result_contents for result_contents, _ in execution_results], should_terminate
@@ -2872,10 +2890,10 @@ def _store_already_approved_approval_requests(
     invocation_session: AgentSession | None,
     visible_approval_requests: Sequence[Content],
     already_approved_requests: Sequence[Content],
+    *,
+    function_call_order: Sequence[Mapping[str, str | None]] | 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
@@ -2888,49 +2906,115 @@ 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],
+        _FUNCTION_CALL_ORDER_KEY: [dict(item) for item in function_call_order]
+        if function_call_order is not None
+        else [
+            {"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
 
 
-def _pop_already_approved_approval_responses(
+def _stage_approval_batch_responses(
     invocation_session: AgentSession | None,
-    approval_response_ids: set[str],
-) -> list[Content]:
-    """Pop already-approved requests for the visible 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
+            if request_id not in stored_responses:
+                stored_responses[request_id] = rebound
+            matched = True
+
+        if not matched:
             remaining_groups.append(raw_group)
             continue
-        raw_requests = group.get("approval_requests")
-        if not isinstance(raw_requests, list):
+
+        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
-        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))
+
+        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):
+                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 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
+    _save_pending_approval_requests(invocation_session, pending_requests)
+    return responses, function_call_order, waiting_requests
 
 
 def _store_pending_mixed_pause_batch(
@@ -3278,11 +3362,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]:
@@ -3335,6 +3420,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:
@@ -4046,30 +4164,91 @@ 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
-    }
-
-    if already_approved_responses := _pop_already_approved_approval_responses(
-        approval_session,
-        explicit_approval_response_ids,
-    ):
-        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(
+            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}
+        batch_responses: list[Content] = []
+        for message in prepared_messages:
+            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 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))
+
+        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 batch_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.
     if not (
diff --git a/python/packages/core/agent_framework/_workflows/_agent_executor.py b/python/packages/core/agent_framework/_workflows/_agent_executor.py
index 6867bd73a05..a56cf2d0331 100644
--- a/python/packages/core/agent_framework/_workflows/_agent_executor.py
+++ b/python/packages/core/agent_framework/_workflows/_agent_executor.py
@@ -342,7 +342,23 @@ 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)
+            )
+        elif (
+            cancelled_request is not None
+            and cancelled_request.type == "function_call"
+            and cancelled_request.call_id is not None
+        ):
+            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/core/test_function_invocation_logic.py b/python/packages/core/tests/core/test_function_invocation_logic.py
index 9dea58bf4c2..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
@@ -3463,6 +3525,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,
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 b3b88359215..cf5e179153c 100644
--- a/python/packages/core/tests/core/test_harness_tool_approval.py
+++ b/python/packages/core/tests/core/test_harness_tool_approval.py
@@ -1502,20 +1502,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(
@@ -1550,6 +1554,241 @@ 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_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"]
+
+
+@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_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_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(
diff --git a/python/packages/core/tests/core/test_tools.py b/python/packages/core/tests/core/test_tools.py
index 89aa73486cc..4b0eb1c81a9 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,100 @@
 # 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_finishes_batch_after_termination(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    executed_calls: list[str] = []
+
+    async def execute(function_call: Content, **_: Any) -> tuple[list[Content], bool]:
+        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)
+
+    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", "second"]
+    assert [group[0].result for group in result_groups] == ["done", "done"]
+
+
 def test_tool_decorator():
     """Test the tool decorator."""
 
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..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,13 +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) -> 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,
         *,
@@ -537,7 +567,11 @@ 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", self._mixed_request_contents())
+                )
+            elif self._parallel_request:
                 response = ChatResponse(
                     messages=Message(
                         "assistant",
@@ -570,7 +604,12 @@ 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=self._mixed_request_contents(),
+                    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 +767,86 @@ 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
+
+
+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