fix: prevent AssistantAgent deadlock on tool call cancellation

python/packages/autogen-agentchat/src/autogen_agentchat/agents/_assistant_agent.py

@@ -1208,9 +1208,12 @@ async def _execute_tool_calls(
                             stream=stream_queue,
                         )
                         for call in function_calls
-                    ]
+                    ],
+                    return_exceptions=True,
                 )
                 # Signal the end of streaming by putting None in the queue.
+                # This MUST be in a finally block to ensure the consumer loop
+                # always terminates, even if a tool call raises (e.g. CancelledError).
                 stream_queue.put_nowait(None)
                 return results
 
@@ -1229,7 +1232,27 @@ async def _execute_tool_calls(
 
             # Wait for all tool calls to complete.
             executed_calls_and_results = await task
-            exec_results = [result for _, result in executed_calls_and_results]
+
+            # Process results, converting any exceptions to error FunctionExecutionResult
+            exec_results: List[FunctionExecutionResult] = []
+            processed_calls_and_results: List[Tuple[FunctionCall, FunctionExecutionResult]] = []
+            for item in executed_calls_and_results:
+                if isinstance(item, BaseException):
+                    # Tool call raised an exception (e.g. CancelledError)
+                    # Create a placeholder FunctionCall and an error result
+                    placeholder_call = FunctionCall(id="error", arguments="{}", name="error")
+                    error_result = FunctionExecutionResult(
+                        content=f"Tool execution error: {type(item).__name__}: {item}",
+                        call_id=placeholder_call.id,
+                        is_error=True,
+                        name=placeholder_call.name,
+                    )
+                    exec_results.append(error_result)
+                    processed_calls_and_results.append((placeholder_call, error_result))
+                else:
+                    call, result = item
+                    exec_results.append(result)
+                    processed_calls_and_results.append((call, result))
 
             # Yield ToolCallExecutionEvent
             tool_call_result_msg = ToolCallExecutionEvent(
@@ -1244,7 +1267,7 @@ async def _execute_tool_calls(
             # STEP 4C: Check for handoff
             handoff_output = cls._check_and_handle_handoff(
                 model_result=current_model_result,
-                executed_calls_and_results=executed_calls_and_results,
+                executed_calls_and_results=processed_calls_and_results,
                 inner_messages=inner_messages,
                 handoffs=handoffs,
                 agent_name=agent_name,
@@ -1315,7 +1338,7 @@ async def _execute_tool_calls(
                 yield reflection_response
         else:
             yield cls._summarize_tool_use(
-                executed_calls_and_results=executed_calls_and_results,
+                executed_calls_and_results=processed_calls_and_results,
                 inner_messages=inner_messages,
                 handoffs=handoffs,
                 tool_call_summary_format=tool_call_summary_format,

python/packages/autogen-agentchat/tests/test_assistant_agent.py

@@ -2816,6 +2816,68 @@ async def test_reset_with_cancellation_token(self) -> None:
         # Context clear should be called
         mock_context.clear.assert_called_once()
 
+    @pytest.mark.asyncio
+    async def test_stream_terminates_after_tool_cancellation(self) -> None:
+        """Regression test: streaming operation must terminate cleanly when
+        a tool call is cancelled mid-flight. Previously, cancellation could
+        leave the stream blocking forever on queue.get().
+
+        See https://github.com/microsoft/autogen/issues/8092"""
+
+        import asyncio
+
+        # A tool that blocks until cancelled
+        async def blocking_tool() -> str:
+            # Wait forever - will only return when cancelled
+            await asyncio.sleep(3600)
+
+        model_client = ReplayChatCompletionClient(
+            [
+                CreateResult(
+                    finish_reason="function_calls",
+                    content=[FunctionCall(id="1", arguments="{}", name="blocking_tool")],
+                    usage=RequestUsage(prompt_tokens=10, completion_tokens=5),
+                    cached=False,
+                ),
+            ],
+            model_info={
+                "function_calling": True,
+                "vision": False,
+                "json_output": False,
+                "family": ModelFamily.GPT_4O,
+                "structured_output": False,
+            },
+        )
+
+        agent = AssistantAgent(
+            name="test_agent",
+            model_client=model_client,
+            tools=[blocking_tool],
+        )
+
+        cancellation_token = CancellationToken()
+
+        # Run the stream in a task so we can cancel it
+        async def consume_stream() -> None:
+            async for event in agent.on_messages_stream(
+                [TextMessage(content="Test", source="user")], cancellation_token
+            ):
+                pass  # We don't care about events, just that the stream terminates
+
+        task = asyncio.create_task(consume_stream())
+
+        # Give the stream time to start and reach the tool call
+        await asyncio.sleep(0.1)
+
+        # Cancel the operation - this should cause the stream to terminate
+        cancellation_token.cancel()
+
+        # The stream must terminate (not hang forever)
+        try:
+            await asyncio.wait_for(task, timeout=2.0)
+        except (asyncio.TimeoutError, asyncio.CancelledError):
+            pytest.fail("Stream did not terminate after cancellation - deadlock detected")
+
 
 class TestAssistantAgentStreamingEdgeCases:
     """Test suite for streaming edge cases and error scenarios."""