fix: preserve concurrent tool result order

src/strands/tools/executors/concurrent.py

@@ -49,14 +49,15 @@ async def _execute(
         stop_event = object()
 
         tasks = []
+        ordered_tool_results: list[list[ToolResult]] = [[] for _ in tool_uses]
         try:
             for task_id, tool_use in enumerate(tool_uses):
                 tasks.append(
                     asyncio.create_task(
                         self._task(
                             agent,
                             tool_use,
-                            tool_results,
+                            ordered_tool_results[task_id],
                             cycle_trace,
                             cycle_span,
                             invocation_state,
@@ -81,6 +82,9 @@ async def _execute(
 
                 yield event
                 task_events[task_id].set()
+
+            for task_results in ordered_tool_results:
+                tool_results.extend(task_results)
         finally:
             for task in tasks:
                 task.cancel()

tests/strands/tools/executors/test_concurrent.py

@@ -1,5 +1,8 @@
+import asyncio
+
 import pytest
 
+import strands
 from strands.hooks import AfterToolCallEvent, BeforeToolCallEvent
 from strands.interrupt import Interrupt
 from strands.tools.executors import ConcurrentToolExecutor
@@ -41,6 +44,35 @@ async def test_concurrent_executor_execute(
     assert tru_results == exp_results
 
 
+@pytest.mark.asyncio
+async def test_concurrent_executor_keeps_tool_result_order(
+    executor, agent, tool_results, cycle_trace, cycle_span, invocation_state, structured_output_context, alist
+):
+    @strands.tool(name="slow_order_tool")
+    async def slow_order_tool():
+        await asyncio.sleep(0.05)
+        return "slow"
+
+    @strands.tool(name="fast_order_tool")
+    async def fast_order_tool():
+        return "fast"
+
+    agent.tool_registry.register_tool(slow_order_tool)
+    agent.tool_registry.register_tool(fast_order_tool)
+
+    tool_uses = [
+        {"name": "slow_order_tool", "toolUseId": "slow-id", "input": {}},
+        {"name": "fast_order_tool", "toolUseId": "fast-id", "input": {}},
+    ]
+    stream = executor._execute(
+        agent, tool_uses, tool_results, cycle_trace, cycle_span, invocation_state, structured_output_context
+    )
+
+    await alist(stream)
+
+    assert [result["toolUseId"] for result in tool_results] == ["slow-id", "fast-id"]
+
+
 @pytest.mark.asyncio
 async def test_concurrent_executor_interrupt(
     executor, agent, tool_results, cycle_trace, cycle_span, invocation_state, structured_output_context, alist