@@ -550,31 +550,29 @@ async def _agent_wrapper(ctx: FunctionInvocationContext, **kwargs: Any) -> str:
             """
             session = ctx.session if propagate_session else None
 
-            # Isolate the child agent from the parent's server-side conversation.
-            # service_session_id would cause the child to send previous_response_id
-            # referencing the parent's pending tool_call, resulting in a 400 error.
-            saved_service_session_id = None
-            if session is not None and session.service_session_id is not None:
-                saved_service_session_id = session.service_session_id
-                session.service_session_id = None
-
-            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:
-                    stream.with_transform_hook(stream_callback)
-                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
-            finally:
-                if session is not None and saved_service_session_id is not None:
-                    session.service_session_id = saved_service_session_id
+            # 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:
+                stream.with_transform_hook(stream_callback)
+            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
 
         return FunctionTool(
             name=tool_name,
 
@@ -1010,7 +1010,7 @@ def capturing_run(*args: Any, **kwargs: Any) -> Any:
 
 
 async def test_chat_agent_as_tool_propagate_session_clears_service_session_id(client: SupportsChatGetResponse) -> None:
-    """Test that propagate_session=True clears service_session_id for the child and restores it after."""
+    """Test that propagate_session=True gives the child a separate session with cleared service_session_id."""
     agent = Agent(client=client, name="SubAgent", description="Sub agent")
     tool = agent.as_tool(propagate_session=True)
 
@@ -1024,9 +1024,11 @@ async def test_chat_agent_as_tool_propagate_session_clears_service_session_id(cl
     def capturing_run(*args: Any, **kwargs: Any) -> Any:
         nonlocal captured_session
         captured_session = kwargs.get("session")
-        # The child should see the same session object but with service_session_id cleared
-        assert captured_session is parent_session
+        # The child gets a different session object with isolated service_session_id
+        assert captured_session is not parent_session
         assert captured_session.service_session_id is None
+        # But shares the same state dict by reference
+        assert captured_session.state is parent_session.state
         assert captured_session.state["data"] == "shared"
         return original_run(*args, **kwargs)
 
@@ -1040,14 +1042,14 @@ def capturing_run(*args: Any, **kwargs: Any) -> Any:
         )
     )
 
-    # After the child finishes, service_session_id is restored
+    # Parent's service_session_id is never mutated
     assert parent_session.service_session_id == "resp_parent_abc123"
 
 
 async def test_chat_agent_as_tool_propagate_session_restores_service_session_id_on_error(
     client: SupportsChatGetResponse,
 ) -> None:
-    """Test that service_session_id is restored even if the child agent raises."""
+    """Test that parent's service_session_id is untouched even if the child agent raises."""
     agent = Agent(client=client, name="SubAgent", description="Sub agent")
     tool = agent.as_tool(propagate_session=True)
 
@@ -1068,12 +1070,12 @@ def failing_run(*args: Any, **kwargs: Any) -> Any:
             )
         )
 
-    # service_session_id must be restored even after failure
+    # Parent's service_session_id is never mutated — child has its own session
     assert parent_session.service_session_id == "resp_parent_xyz789"
 
 
 async def test_chat_agent_as_tool_propagate_session_no_service_session_id(client: SupportsChatGetResponse) -> None:
-    """Test that when service_session_id is None, no save/restore is needed."""
+    """Test that child setting service_session_id does not leak back to the parent."""
     agent = Agent(client=client, name="SubAgent", description="Sub agent")
     tool = agent.as_tool(propagate_session=True)
 
@@ -1087,6 +1089,8 @@ def capturing_run(*args: Any, **kwargs: Any) -> Any:
         nonlocal captured_session
         captured_session = kwargs.get("session")
         assert captured_session.service_session_id is None
+        # Simulate the child's run populating service_session_id
+        captured_session.service_session_id = "resp_child_leaked"
         return original_run(*args, **kwargs)
 
     agent.run = capturing_run  # type: ignore[assignment, method-assign]
@@ -1099,6 +1103,7 @@ def capturing_run(*args: Any, **kwargs: Any) -> Any:
         )
     )
 
+    # The child's service_session_id must not leak back to the parent
     assert parent_session.service_session_id is None