fix: Fix double execution

Author: Murat Kaan Meral

src/strands/multiagent/graph.py

@@ -673,7 +673,8 @@ async def _execute_node(self, node: GraphNode, invocation_state: dict[str, Any])
             # Execute with timeout protection and stream events
             try:
                 if isinstance(node.executor, MultiAgentBase):
-                    # For nested multi-agent systems, stream their events
+                    # For nested multi-agent systems, stream their events and collect result
+                    multi_agent_result = None
                     if self.node_timeout is not None:
                         # Implement timeout for async generator streaming
                         async for event in self._stream_with_timeout(
@@ -684,14 +685,22 @@ async def _execute_node(self, node: GraphNode, invocation_state: dict[str, Any])
                             # Forward nested multi-agent events with node context
                             wrapped_event = MultiAgentNodeStreamEvent(node.node_id, event)
                             yield wrapped_event.as_dict()
+                            # Capture the final result event
+                            if "result" in event:
+                                multi_agent_result = event["result"]
                     else:
                         async for event in node.executor.stream_async(node_input, invocation_state):
                             # Forward nested multi-agent events with node context
                             wrapped_event = MultiAgentNodeStreamEvent(node.node_id, event)
                             yield wrapped_event.as_dict()
+                            # Capture the final result event
+                            if "result" in event:
+                                multi_agent_result = event["result"]
+
+                    # Use the captured result from streaming (no double execution)
+                    if multi_agent_result is None:
+                        raise ValueError(f"Node '{node.node_id}' did not produce a result event")
 
-                    # Get the final result for metrics
-                    multi_agent_result = await node.executor.invoke_async(node_input, invocation_state)
                     node_result = NodeResult(
                         result=multi_agent_result,
                         execution_time=multi_agent_result.execution_time,
@@ -702,7 +711,8 @@ async def _execute_node(self, node: GraphNode, invocation_state: dict[str, Any])
                     )
 
                 elif isinstance(node.executor, Agent):
-                    # For agents, stream their events
+                    # For agents, stream their events and collect result
+                    agent_response = None
                     if self.node_timeout is not None:
                         # Implement timeout for async generator streaming
                         async for event in self._stream_with_timeout(
@@ -713,14 +723,22 @@ async def _execute_node(self, node: GraphNode, invocation_state: dict[str, Any])
                             # Forward agent events with node context
                             wrapped_event = MultiAgentNodeStreamEvent(node.node_id, event)
                             yield wrapped_event.as_dict()
+                            # Capture the final result event
+                            if "result" in event:
+                                agent_response = event["result"]
                     else:
                         async for event in node.executor.stream_async(node_input, **invocation_state):
                             # Forward agent events with node context
                             wrapped_event = MultiAgentNodeStreamEvent(node.node_id, event)
                             yield wrapped_event.as_dict()
+                            # Capture the final result event
+                            if "result" in event:
+                                agent_response = event["result"]
+
+                    # Use the captured result from streaming (no double execution)
+                    if agent_response is None:
+                        raise ValueError(f"Node '{node.node_id}' did not produce a result event")
 
-                    # Get the final result for metrics
-                    agent_response = await node.executor.invoke_async(node_input, **invocation_state)
                     usage = Usage(inputTokens=0, outputTokens=0, totalTokens=0)
                     metrics = Metrics(latencyMs=0)
                     if hasattr(agent_response, "metrics") and agent_response.metrics:

src/strands/multiagent/swarm.py

@@ -350,6 +350,19 @@ async def stream_async(
             finally:
                 self.state.execution_time = round((time.time() - start_time) * 1000)
 
+    async def _stream_with_timeout(
+        self, async_generator: AsyncIterator[dict[str, Any]], timeout: float, timeout_message: str
+    ) -> AsyncIterator[dict[str, Any]]:
+        """Wrap an async generator with timeout functionality."""
+        while True:
+            try:
+                event = await asyncio.wait_for(async_generator.__anext__(), timeout=timeout)
+                yield event
+            except StopAsyncIteration:
+                break
+            except asyncio.TimeoutError:
+                raise Exception(timeout_message) from None
+
     def _setup_swarm(self, nodes: list[Agent]) -> None:
         """Initialize swarm configuration."""
         # Validate nodes before setup
@@ -610,9 +623,17 @@ async def _execute_swarm(self, invocation_state: dict[str, Any]) -> AsyncIterato
 
                 # Execute node with timeout protection
                 try:
-                    # For now, execute without timeout for async generators
-                    # TODO: Implement proper timeout for async generators if needed
-                    async for event in self._execute_node(current_node, self.state.task, invocation_state):
+                    # Execute with timeout wrapper for async generator streaming
+                    node_stream = (
+                        self._stream_with_timeout(
+                            self._execute_node(current_node, self.state.task, invocation_state),
+                            self.node_timeout,
+                            f"Node '{current_node.node_id}' execution timed out after {self.node_timeout}s",
+                        )
+                        if self.node_timeout is not None
+                        else self._execute_node(current_node, self.state.task, invocation_state)
+                    )
+                    async for event in node_stream:
                         yield event
 
                     self.state.node_history.append(current_node)
@@ -639,17 +660,16 @@ async def _execute_swarm(self, invocation_state: dict[str, Any]) -> AsyncIterato
                         self.state.completion_status = Status.COMPLETED
                         break
 
-                except asyncio.TimeoutError:
-                    logger.exception(
-                        "node=<%s>, timeout=<%s>s | node execution timed out",
-                        current_node.node_id,
-                        self.node_timeout,
-                    )
-                    self.state.completion_status = Status.FAILED
-                    break
-
-                except Exception:
-                    logger.exception("node=<%s> | node execution failed", current_node.node_id)
+                except Exception as e:
+                    # Check if this is a timeout exception
+                    if "timed out after" in str(e):
+                        logger.exception(
+                            "node=<%s>, timeout=<%s>s | node execution timed out",
+                            current_node.node_id,
+                            self.node_timeout,
+                        )
+                    else:
+                        logger.exception("node=<%s> | node execution failed", current_node.node_id)
                     self.state.completion_status = Status.FAILED
                     break
 
@@ -691,14 +711,20 @@ async def _execute_node(
             # Execute node with streaming
             node.reset_executor_state()
 
-            # Stream agent events with node context
+            # Stream agent events with node context and capture final result
+            result = None
             async for event in node.executor.stream_async(node_input, **invocation_state):
                 # Forward agent events with node context
                 wrapped_event = MultiAgentNodeStreamEvent(node_name, event)
                 yield wrapped_event.as_dict()
+                # Capture the final result event
+                if "result" in event:
+                    result = event["result"]
+
+            # Use the captured result from streaming to avoid double execution
+            if result is None:
+                raise ValueError(f"Node '{node_name}' did not produce a result event")
 
-            # Get the final result for metrics (we need to call invoke_async again for the result)
-            result = await node.executor.invoke_async(node_input, **invocation_state)
             execution_time = round((time.time() - start_time) * 1000)
 
             # Create NodeResult