Python: Fail closed on unresolved agent-tool approvals

docs/specs/004-python-function-calling-loop.md

@@ -500,6 +500,15 @@ that manually replay messages own the equivalent rule: do not resend an approval
   approval `Message`, approval `Content`, or an earlier returned response.
 - Approval-time `UserInputRequiredException` and `MiddlewareTermination` return immediately without another model
   call.
+- `Agent.as_tool()` keeps child function approvals inside the delegated invocation. A child
+  `ToolApprovalMiddleware` may resolve them through runtime `auto_approval_rules`; any unresolved child function
+  approval does not enter the caller's approval state or model transcript. An approval-only response fails the
+  agent-tool invocation; a mixed response preserves its non-approval user-input requests while discarding the child
+  approval continuation. Interactive, delayed, or durable approval belongs in a workflow. When
+  `propagate_session=True`, child application-state changes merge back into the parent while framework approval and
+  invocation-budget state remain isolated, including approval queues stored under custom child middleware
+  `source_id` values. Parent and child `ToolApprovalMiddleware` instances must use distinct `source_id` values; an
+  overlap fails before the child runs.
 
 ### Approval control content
 
@@ -578,6 +587,7 @@ that manually replay messages own the equivalent rule: do not resend an approval
 | Mixed approved/rejected batch | Every call gets one correctly correlated terminal result. | `packages/core/tests/core/test_function_invocation_logic.py::test_rejected_approval` |
 | Persisted approval replay | Resume executes with the prior call available. | `test_persisted_approval_messages_replay_correctly` |
 | Hosted approval pass-through | Hosted requests/responses are bound to the recorded provider request and are not processed as local calls. | `test_hosted_tool_approval_response`, `test_hosted_mcp_approval_response_passthrough`, `test_session_approval_binding_reconstructs_hosted_response`, `test_mixed_local_and_hosted_approval_flow` |
+| Agent-tool child approval | A child `ToolApprovalMiddleware` can auto-approve runtime tool requests inside one delegated invocation; unresolved child approvals execute nothing and do not enter the caller's approval state or dispatch a same-named parent tool, including with propagated application state. Mixed batches preserve non-approval user-input requests. Shared parent and child state rejects overlapping `ToolApprovalMiddleware.source_id` values before running the child, and custom child approval queues do not leak into later delegations. | `packages/core/tests/core/test_agents.py::test_chat_agent_as_tool_auto_approves_child_tool_with_middleware`, `test_chat_agent_as_tool_fails_closed_for_unresolved_child_approval`, `test_chat_agent_as_tool_child_approval_does_not_dispatch_same_named_parent_tool`, `test_chat_agent_as_tool_preserves_non_approval_requests_from_mixed_child_batch`, `test_chat_agent_as_tool_shared_session_requires_distinct_tool_approval_source_ids`, `test_chat_agent_as_tool_approved_delegation_does_not_confuse_framework_approval_state`, `test_chat_agent_as_tool_does_not_restore_custom_approval_queue_on_fresh_delegation` |
 | Approval-time user input | Every user-input request from one approved execution returns in order with assistant role and no extra model call; the execution consumes one call-budget unit. | `packages/core/tests/core/test_harness_tool_approval.py::test_approval_resume_returns_all_user_input_requests_without_another_model_call`, `packages/core/tests/core/test_function_invocation_logic.py::test_approval_resume_user_input_counts_toward_function_call_budget` |
 | Mixed terminal result and follow-up input | Completed siblings remain tool-role while only follow-up input requests use assistant-role messages/updates. | `packages/core/tests/core/test_function_invocation_logic.py::test_approval_resume_separates_terminal_results_from_follow_up_requests`, `packages/openai/tests/openai/test_openai_chat_completion_client.py::test_mixed_approval_resume_roles_serialize_function_result_as_tool`, `packages/core/tests/core/test_harness_tool_approval.py::test_dynamic_policy_approval_partitions_safe_sibling_result_roles` |
 | Approval-time middleware termination | Terminal result returns with no extra model call in either response mode. | `packages/core/tests/core/test_function_invocation_logic.py::test_approval_resume_honors_middleware_termination` |

python/packages/core/agent_framework/_agents.py

@@ -63,7 +63,12 @@
     map_chat_to_agent_update,
     normalize_messages,
 )
-from .exceptions import AgentInvalidRequestException, AgentInvalidResponseException, UserInputRequiredException
+from .exceptions import (
+    AgentInvalidRequestException,
+    AgentInvalidResponseException,
+    ToolExecutionException,
+    UserInputRequiredException,
+)
 from .observability import AgentTelemetryLayer
 
 if sys.version_info >= (3, 13):
@@ -95,6 +100,37 @@
 # nested ``agent.run()`` (fresh options, its own session) keeps its own turn,
 # and nothing leaks into the caller's context while a stream is paused.
 _LOOP_ITERATION_TOKEN_KEY = "_agent_loop_iteration"  # nosec B105 - a context-options key, not a credential  # ruff: ignore[hardcoded-password-string]
+_DELEGATED_STATE_MISSING = object()
+
+
+def _tool_approval_source_ids(middleware: Sequence[MiddlewareTypes] | None) -> frozenset[str]:
+    """Return session-state keys owned by ToolApprovalMiddleware instances."""
+    from ._harness._tool_approval import ToolApprovalMiddleware
+
+    return frozenset(item.source_id for item in middleware or () if isinstance(item, ToolApprovalMiddleware))
+
+
+def _merge_delegated_session_state(
+    parent_state: MutableMapping[str, Any],
+    initial_child_state: Mapping[str, Any],
+    final_child_state: Mapping[str, Any],
+    *,
+    excluded_keys: frozenset[str],
+) -> None:
+    """Merge child application-state changes without copying framework continuation state."""
+    for key, initial_value in initial_child_state.items():
+        if key in excluded_keys or key in final_child_state:
+            continue
+        if parent_state.get(key, _DELEGATED_STATE_MISSING) is initial_value:
+            parent_state.pop(key, None)
+
+    for key, final_value in final_child_state.items():
+        if key in excluded_keys:
+            continue
+        initial_value = initial_child_state.get(key, _DELEGATED_STATE_MISSING)
+        if initial_value is _DELEGATED_STATE_MISSING or final_value is not initial_value:
+            parent_state[key] = final_value
+
 
 if TYPE_CHECKING:
     ResponseModelBoundT = TypeVar("ResponseModelBoundT", bound=BaseModel)
@@ -624,12 +660,26 @@ def as_tool(
             approval_mode: Whether this delegated tool requires approval before execution.
             stream_callback: Optional callback for streaming responses. If provided, uses run(..., stream=True).
             propagate_session: If True, the parent agent's session is forwarded
-                to this sub-agent's ``run()`` call so both agents share the
-                same session. Defaults to False.
+                to this sub-agent's ``run()`` call. Application-state changes
+                propagate back to the parent, while framework approval continuation
+                state remains isolated. Defaults to False. The sub-agent always
+                receives an AgentSession so session-backed middleware can run.
+                When False, that session is private to this invocation.
 
         Returns:
             A FunctionTool that can be used as a tool by other agents.
 
+        Note:
+            Child function approvals are not propagated into the calling agent.
+            Configure ToolApprovalMiddleware with runtime auto-approval rules on
+            the child for immediate policy decisions. Use a workflow when approval
+            is interactive, delayed, or durable.
+
+            When parent and child both use ToolApprovalMiddleware with
+            ``propagate_session=True``, configure distinct middleware ``source_id``
+            values. The delegated call raises ToolExecutionException before running
+            the child when their shared session-state keys overlap.
+
         Examples:
             .. code-block:: python
 
@@ -676,39 +726,113 @@ async def _agent_wrapper(ctx: FunctionInvocationContext, **kwargs: Any) -> str:
                 ctx: the function invocation context used
                 **kwargs: only used to dynamically load the argument that is defined for this tool.
             """
-            session = ctx.session if propagate_session else None
-
-            # Create a child session that shares the parent's state dict but has
-            # an isolated service_session_id. This avoids mutating the parent
-            # session in-place, which would race under concurrent asyncio.gather
-            # tool invocations sharing the same session.
-            if session is not None:
-                child_session = AgentSession(session_id=session.session_id)
-                child_session.state = session.state  # shared by reference
-                child_session.service_session_id = None
-                session = child_session
-
-            stream = self.run(
-                str(kwargs.get(arg_name, "")),
-                stream=True,
-                session=session,
-                function_invocation_kwargs=dict(ctx.kwargs),
-            )
-            if stream_callback is not None:
-                # The callback is a host-facing observer: feed it the *released*
-                # updates by consuming the stream, never by registering a transform
-                # hook on it. Hooks can end up applied to buffered content ahead of an
-                # egress gate's verdict (see ResponseStream.buffered_and_gated), so a
-                # hook-registered observer could see denied or unredacted content.
-                async for update in stream:
-                    callback_result = stream_callback(update)
-                    if isawaitable(callback_result):
-                        await callback_result
-            final_response = await stream.get_final_response()
-            if final_response.user_input_requests:
-                raise UserInputRequiredException(contents=final_response.user_input_requests)
-            # TODO(Copilot): update once #4331 merges
-            return final_response.text
+            parent_session = ctx.session
+            session = AgentSession()
+            child_approval_source_ids = _tool_approval_source_ids(self.middleware)
+
+            if propagate_session and parent_session is not None:
+                from ._tools import _PARENT_TOOL_APPROVAL_SOURCE_IDS_CONTEXT_KEY  # pyright: ignore[reportPrivateUsage]
+
+                raw_parent_approval_source_ids = ctx.metadata.get(_PARENT_TOOL_APPROVAL_SOURCE_IDS_CONTEXT_KEY)
+                parent_approval_source_ids: frozenset[str]
+                parent_approval_source_ids = (
+                    cast("frozenset[str]", raw_parent_approval_source_ids)
+                    if isinstance(raw_parent_approval_source_ids, frozenset)
+                    else frozenset()
+                )
+                overlapping_source_ids = child_approval_source_ids.intersection(parent_approval_source_ids)
+                if overlapping_source_ids:
+                    formatted_source_ids = ", ".join(repr(source_id) for source_id in sorted(overlapping_source_ids))
+                    raise ToolExecutionException(
+                        f"Agent tool {tool_name!r} cannot share its parent session because parent and child "
+                        f"ToolApprovalMiddleware instances use the same source_id: {formatted_source_ids}. "
+                        "Configure distinct source_id values or set propagate_session=False."
+                    )
+
+            parent_state: MutableMapping[str, Any] | None = None
+            initial_child_state: dict[str, Any] | None = None
+            excluded_state_keys: frozenset[str] = frozenset()
+
+            # Propagate application state through a child-owned copy. Framework
+            # approval continuation state stays isolated so an unresolved child
+            # request can never become pending authority in the parent session.
+            if propagate_session and parent_session is not None:
+                from ._tools import (
+                    _FUNCTION_INVOCATION_BUDGET_STATE_KEY,  # pyright: ignore[reportPrivateUsage]
+                    _FUNCTION_RESULT_PAYLOAD_BUDGET_STATE_KEY,  # pyright: ignore[reportPrivateUsage]
+                    _TOOL_APPROVAL_STATE_KEY,  # pyright: ignore[reportPrivateUsage]
+                )
+
+                excluded_state_keys = frozenset({
+                    _TOOL_APPROVAL_STATE_KEY,
+                    _FUNCTION_INVOCATION_BUDGET_STATE_KEY,
+                    _FUNCTION_RESULT_PAYLOAD_BUDGET_STATE_KEY,
+                    *child_approval_source_ids,
+                })
+                parent_state = parent_session.state
+                child_state = {key: value for key, value in parent_state.items() if key not in excluded_state_keys}
+                initial_child_state = dict(child_state)
+                session = AgentSession(session_id=parent_session.session_id)
+                session.state = child_state
+
+            try:
+                stream = self.run(
+                    str(kwargs.get(arg_name, "")),
+                    stream=True,
+                    session=session,
+                    function_invocation_kwargs=dict(ctx.kwargs),
+                )
+                if stream_callback is not None:
+                    # The callback is a host-facing observer: feed it the *released*
+                    # updates by consuming the stream, never by registering a transform
+                    # hook on it. Hooks can end up applied to buffered content ahead of an
+                    # egress gate's verdict (see ResponseStream.buffered_and_gated), so a
+                    # hook-registered observer could see denied or unredacted content.
+                    async for update in stream:
+                        callback_result = stream_callback(update)
+                        if isawaitable(callback_result):
+                            await callback_result
+                final_response = await stream.get_final_response()
+                approval_requests = [
+                    request
+                    for request in final_response.user_input_requests
+                    if request.type == "function_approval_request"
+                ]
+                other_input_requests = [
+                    request
+                    for request in final_response.user_input_requests
+                    if request.type != "function_approval_request"
+                ]
+                if approval_requests:
+                    requested_tools = sorted(
+                        {
+                            request.function_call.name or "<unknown>"
+                            for request in approval_requests
+                            if request.function_call is not None
+                        }
+                        or {"<unknown>"}
+                    )
+                    approval_error = (
+                        f"Agent tool {tool_name!r} cannot continue because its sub-agent requested approval for "
+                        f"{', '.join(requested_tools)}. Configure ToolApprovalMiddleware with auto_approval_rules on "
+                        "the sub-agent for immediate policy decisions. Use a workflow for interactive, delayed, or "
+                        "durable approval."
+                    )
+                    if other_input_requests:
+                        raise UserInputRequiredException(contents=other_input_requests, message=approval_error)
+                    raise ToolExecutionException(approval_error)
+                if other_input_requests:
+                    raise UserInputRequiredException(contents=other_input_requests)
+                # TODO(Copilot): update once #4331 merges
+                return final_response.text
+            finally:
+                if parent_state is not None and initial_child_state is not None:
+                    _merge_delegated_session_state(
+                        parent_state,
+                        initial_child_state,
+                        session.state,
+                        excluded_keys=excluded_state_keys,
+                    )
 
         from ._tools import FunctionTool
 
@@ -1450,6 +1574,7 @@ async def _prepare_run_context(
 
         agent_name = self._get_agent_name()
         from ._mcp import MCPTool
+        from ._tools import _PARENT_TOOL_APPROVAL_SOURCE_IDS_CONTEXT_KEY  # pyright: ignore[reportPrivateUsage]
 
         base_tools = _normalize_tools(chat_options.pop("tools", None))
         mcp_duplicate_message = "Tool names must be unique. Consider setting `tool_name_prefix` on the MCPTool."
@@ -1494,6 +1619,10 @@ async def _prepare_run_context(
                 duplicate_error_message=mcp_duplicate_message,
             )
 
+        additional_function_arguments[_PARENT_TOOL_APPROVAL_SOURCE_IDS_CONTEXT_KEY] = _tool_approval_source_ids(
+            self.middleware
+        )
+
         model = opts.pop("model", None)
 
         # Build options dict from run() options merged with provided options

python/packages/core/agent_framework/_tools.py

@@ -106,6 +106,7 @@ def _generate_function_call_occurrence_id() -> str:
 SHELL_TOOL_KIND_VALUE: Final[str] = "shell"
 _TOOL_APPROVAL_STATE_KEY: Final[str] = "tool_approval"
 _APPROVAL_SESSION_IS_AUTHORITATIVE_KEY: Final[str] = "_approval_session_is_authoritative"
+_PARENT_TOOL_APPROVAL_SOURCE_IDS_CONTEXT_KEY: Final[str] = "_parent_tool_approval_source_ids"
 
 
 def _has_authoritative_approval_session(invocation_session: AgentSession | None) -> bool:
@@ -2068,8 +2069,21 @@ async def _auto_invoke_function(
     runtime_kwargs: dict[str, Any] = {
         key: value
         for key, value in (custom_args or {}).items()
-        if key not in {"_function_middleware_pipeline", "middleware", "conversation_id"}
+        if key
+        not in {
+            "_function_middleware_pipeline",
+            "middleware",
+            "conversation_id",
+            _PARENT_TOOL_APPROVAL_SOURCE_IDS_CONTEXT_KEY,
+        }
     }
+    raw_parent_approval_source_ids = (custom_args or {}).get(_PARENT_TOOL_APPROVAL_SOURCE_IDS_CONTEXT_KEY)
+    parent_approval_source_ids: frozenset[str]
+    parent_approval_source_ids = (
+        cast("frozenset[str]", raw_parent_approval_source_ids)
+        if isinstance(raw_parent_approval_source_ids, frozenset)
+        else frozenset()
+    )
     if invocation_session is not None:
         runtime_kwargs["session"] = invocation_session
     args = dict(parsed_args)
@@ -2088,6 +2102,7 @@ async def _auto_invoke_function(
                     kwargs=runtime_kwargs.copy(),
                     tools=live_tools,
                 )
+                direct_context.metadata[_PARENT_TOOL_APPROVAL_SOURCE_IDS_CONTEXT_KEY] = parent_approval_source_ids
                 if host_payload_budget is not None:
                     direct_context.metadata[_FUNCTION_RESULT_PAYLOAD_BUDGET_CONTEXT_KEY] = host_payload_budget
             function_result = await tool.invoke(
@@ -2126,6 +2141,7 @@ async def _auto_invoke_function(
         kwargs=runtime_kwargs.copy(),
         tools=live_tools,
     )
+    middleware_context.metadata[_PARENT_TOOL_APPROVAL_SOURCE_IDS_CONTEXT_KEY] = parent_approval_source_ids
     if host_payload_budget is not None:
         middleware_context.metadata[_FUNCTION_RESULT_PAYLOAD_BUDGET_CONTEXT_KEY] = host_payload_budget
     middleware_context.metadata[_AUTO_ARGUMENT_PREPARATION_CONTEXT_KEY] = True
@@ -3061,9 +3077,7 @@ def _stage_approval_batch_responses(
         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
+                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]