chore: hotfix update for crewAI version to 0.203.2

src/crewai/__init__.py

@@ -40,7 +40,7 @@ def _suppress_pydantic_deprecation_warnings() -> None:
 
 _suppress_pydantic_deprecation_warnings()
 
-__version__ = "0.203.1"
+__version__ = "0.203.2"
 _telemetry_submitted = False
 
 

src/crewai/events/listeners/tracing/utils.py

@@ -358,8 +358,7 @@ def prompt_user_for_trace_viewing(timeout_seconds: int = 20) -> bool:
             try:
                 response = input().strip().lower()
                 result[0] = response in ["y", "yes"]
-            except (EOFError, KeyboardInterrupt, OSError, LookupError):
-                # Handle all input-related errors silently
+            except (EOFError, KeyboardInterrupt):
                 result[0] = False
 
         input_thread = threading.Thread(target=get_input, daemon=True)
@@ -372,7 +371,6 @@ def prompt_user_for_trace_viewing(timeout_seconds: int = 20) -> bool:
         return result[0]
 
     except Exception:
-        # Suppress any warnings or errors and assume "no"
         return False
 
 

src/crewai/flow/flow.py

@@ -31,7 +31,7 @@ from crewai.flow.flow_visualizer import plot_flow
 from crewai.flow.persistence.base import FlowPersistence
 from crewai.flow.types import FlowExecutionData
 from crewai.flow.utils import get_possible_return_constants
-from crewai.utilities.printer import Printer, PrinterColor
+from crewai.utilities.printer import Printer
 
 logger = logging.getLogger(__name__)
 
@@ -105,7 +105,7 @@ def start(condition: str | dict | Callable | None = None) -> Callable:
     condition : Optional[Union[str, dict, Callable]], optional
         Defines when the start method should execute. Can be:
         - str: Name of a method that triggers this start
-        - dict: Result from or_() or and_(), including nested conditions
+        - dict: Contains "type" ("AND"/"OR") and "methods" (list of triggers)
         - Callable: A method reference that triggers this start
         Default is None, meaning unconditional start.
 
@@ -140,18 +140,13 @@ def start(condition: str | dict | Callable | None = None) -> Callable:
             if isinstance(condition, str):
                 func.__trigger_methods__ = [condition]
                 func.__condition_type__ = "OR"
-            elif isinstance(condition, dict) and "type" in condition:
-                if "conditions" in condition:
-                    func.__trigger_condition__ = condition
-                    func.__trigger_methods__ = _extract_all_methods(condition)
-                    func.__condition_type__ = condition["type"]
-                elif "methods" in condition:
-                    func.__trigger_methods__ = condition["methods"]
-                    func.__condition_type__ = condition["type"]
-                else:
-                    raise ValueError(
-                        "Condition dict must contain 'conditions' or 'methods'"
-                    )
+            elif (
+                isinstance(condition, dict)
+                and "type" in condition
+                and "methods" in condition
+            ):
+                func.__trigger_methods__ = condition["methods"]
+                func.__condition_type__ = condition["type"]
             elif callable(condition) and hasattr(condition, "__name__"):
                 func.__trigger_methods__ = [condition.__name__]
                 func.__condition_type__ = "OR"
@@ -177,7 +172,7 @@ def listen(condition: str | dict | Callable) -> Callable:
     condition : Union[str, dict, Callable]
         Specifies when the listener should execute. Can be:
         - str: Name of a method that triggers this listener
-        - dict: Result from or_() or and_(), including nested conditions
+        - dict: Contains "type" ("AND"/"OR") and "methods" (list of triggers)
         - Callable: A method reference that triggers this listener
 
     Returns
@@ -205,18 +200,13 @@ def listen(condition: str | dict | Callable) -> Callable:
         if isinstance(condition, str):
             func.__trigger_methods__ = [condition]
             func.__condition_type__ = "OR"
-        elif isinstance(condition, dict) and "type" in condition:
-            if "conditions" in condition:
-                func.__trigger_condition__ = condition
-                func.__trigger_methods__ = _extract_all_methods(condition)
-                func.__condition_type__ = condition["type"]
-            elif "methods" in condition:
-                func.__trigger_methods__ = condition["methods"]
-                func.__condition_type__ = condition["type"]
-            else:
-                raise ValueError(
-                    "Condition dict must contain 'conditions' or 'methods'"
-                )
+        elif (
+            isinstance(condition, dict)
+            and "type" in condition
+            and "methods" in condition
+        ):
+            func.__trigger_methods__ = condition["methods"]
+            func.__condition_type__ = condition["type"]
         elif callable(condition) and hasattr(condition, "__name__"):
             func.__trigger_methods__ = [condition.__name__]
             func.__condition_type__ = "OR"
@@ -243,7 +233,7 @@ def router(condition: str | dict | Callable) -> Callable:
     condition : Union[str, dict, Callable]
         Specifies when the router should execute. Can be:
         - str: Name of a method that triggers this router
-        - dict: Result from or_() or and_(), including nested conditions
+        - dict: Contains "type" ("AND"/"OR") and "methods" (list of triggers)
         - Callable: A method reference that triggers this router
 
     Returns
@@ -276,18 +266,13 @@ def router(condition: str | dict | Callable) -> Callable:
         if isinstance(condition, str):
             func.__trigger_methods__ = [condition]
             func.__condition_type__ = "OR"
-        elif isinstance(condition, dict) and "type" in condition:
-            if "conditions" in condition:
-                func.__trigger_condition__ = condition
-                func.__trigger_methods__ = _extract_all_methods(condition)
-                func.__condition_type__ = condition["type"]
-            elif "methods" in condition:
-                func.__trigger_methods__ = condition["methods"]
-                func.__condition_type__ = condition["type"]
-            else:
-                raise ValueError(
-                    "Condition dict must contain 'conditions' or 'methods'"
-                )
+        elif (
+            isinstance(condition, dict)
+            and "type" in condition
+            and "methods" in condition
+        ):
+            func.__trigger_methods__ = condition["methods"]
+            func.__condition_type__ = condition["type"]
         elif callable(condition) and hasattr(condition, "__name__"):
             func.__trigger_methods__ = [condition.__name__]
             func.__condition_type__ = "OR"
@@ -313,15 +298,14 @@ def or_(*conditions: str | dict | Callable) -> dict:
     *conditions : Union[str, dict, Callable]
         Variable number of conditions that can be:
         - str: Method names
-        - dict: Existing condition dictionaries (nested conditions)
+        - dict: Existing condition dictionaries
         - Callable: Method references
 
     Returns
     -------
     dict
         A condition dictionary with format:
-        {"type": "OR", "conditions": list_of_conditions}
-        where each condition can be a string (method name) or a nested dict
+        {"type": "OR", "methods": list_of_method_names}
 
     Raises
     ------
@@ -333,22 +317,18 @@ def or_(*conditions: str | dict | Callable) -> dict:
     >>> @listen(or_("success", "timeout"))
     >>> def handle_completion(self):
     ...     pass
-
-    >>> @listen(or_(and_("step1", "step2"), "step3"))
-    >>> def handle_nested(self):
-    ...     pass
     """
-    processed_conditions: list[str | dict[str, Any]] = []
+    methods = []
     for condition in conditions:
-        if isinstance(condition, dict):
-            processed_conditions.append(condition)
+        if isinstance(condition, dict) and "methods" in condition:
+            methods.extend(condition["methods"])
         elif isinstance(condition, str):
-            processed_conditions.append(condition)
+            methods.append(condition)
         elif callable(condition):
-            processed_conditions.append(getattr(condition, "__name__", repr(condition)))
+            methods.append(getattr(condition, "__name__", repr(condition)))
         else:
             raise ValueError("Invalid condition in or_()")
-    return {"type": "OR", "conditions": processed_conditions}
+    return {"type": "OR", "methods": methods}
 
 
 def and_(*conditions: str | dict | Callable) -> dict:
@@ -364,15 +344,14 @@ def and_(*conditions: str | dict | Callable) -> dict:
     *conditions : Union[str, dict, Callable]
         Variable number of conditions that can be:
         - str: Method names
-        - dict: Existing condition dictionaries (nested conditions)
+        - dict: Existing condition dictionaries
         - Callable: Method references
 
     Returns
     -------
     dict
         A condition dictionary with format:
-        {"type": "AND", "conditions": list_of_conditions}
-        where each condition can be a string (method name) or a nested dict
+        {"type": "AND", "methods": list_of_method_names}
 
     Raises
     ------
@@ -384,69 +363,18 @@ def and_(*conditions: str | dict | Callable) -> dict:
     >>> @listen(and_("validated", "processed"))
     >>> def handle_complete_data(self):
     ...     pass
-
-    >>> @listen(and_(or_("step1", "step2"), "step3"))
-    >>> def handle_nested(self):
-    ...     pass
     """
-    processed_conditions: list[str | dict[str, Any]] = []
+    methods = []
     for condition in conditions:
-        if isinstance(condition, dict):
-            processed_conditions.append(condition)
+        if isinstance(condition, dict) and "methods" in condition:
+            methods.extend(condition["methods"])
         elif isinstance(condition, str):
-            processed_conditions.append(condition)
+            methods.append(condition)
         elif callable(condition):
-            processed_conditions.append(getattr(condition, "__name__", repr(condition)))
+            methods.append(getattr(condition, "__name__", repr(condition)))
         else:
             raise ValueError("Invalid condition in and_()")
-    return {"type": "AND", "conditions": processed_conditions}
-
-
-def _normalize_condition(condition: str | dict | list) -> dict:
-    """Normalize a condition to standard format with 'conditions' key.
-
-    Args:
-        condition: Can be a string (method name), dict (condition), or list
-
-    Returns:
-        Normalized dict with 'type' and 'conditions' keys
-    """
-    if isinstance(condition, str):
-        return {"type": "OR", "conditions": [condition]}
-    if isinstance(condition, dict):
-        if "conditions" in condition:
-            return condition
-        if "methods" in condition:
-            return {"type": condition["type"], "conditions": condition["methods"]}
-        return condition
-    if isinstance(condition, list):
-        return {"type": "OR", "conditions": condition}
-    return {"type": "OR", "conditions": [condition]}
-
-
-def _extract_all_methods(condition: str | dict | list) -> list[str]:
-    """Extract all method names from a condition (including nested).
-
-    Args:
-        condition: Can be a string, dict, or list
-
-    Returns:
-        List of all method names in the condition tree
-    """
-    if isinstance(condition, str):
-        return [condition]
-    if isinstance(condition, dict):
-        normalized = _normalize_condition(condition)
-        methods = []
-        for sub_cond in normalized.get("conditions", []):
-            methods.extend(_extract_all_methods(sub_cond))
-        return methods
-    if isinstance(condition, list):
-        methods = []
-        for item in condition:
-            methods.extend(_extract_all_methods(item))
-        return methods
-    return []
+    return {"type": "AND", "methods": methods}
 
 
 class FlowMeta(type):
@@ -474,10 +402,7 @@ class FlowMeta(type):
                 if hasattr(attr_value, "__trigger_methods__"):
                     methods = attr_value.__trigger_methods__
                     condition_type = getattr(attr_value, "__condition_type__", "OR")
-                    if hasattr(attr_value, "__trigger_condition__"):
-                        listeners[attr_name] = attr_value.__trigger_condition__
-                    else:
-                        listeners[attr_name] = (condition_type, methods)
+                    listeners[attr_name] = (condition_type, methods)
 
                     if (
                         hasattr(attr_value, "__is_router__")
@@ -897,7 +822,6 @@ class Flow(Generic[T], metaclass=FlowMeta):
                 # Clear completed methods and outputs for a fresh start
                 self._completed_methods.clear()
                 self._method_outputs.clear()
-                self._pending_and_listeners.clear()
             else:
                 # We're restoring from persistence, set the flag
                 self._is_execution_resuming = True
@@ -1162,16 +1086,10 @@ class Flow(Generic[T], metaclass=FlowMeta):
                     for method_name in self._start_methods:
                         # Check if this start method is triggered by the current trigger
                         if method_name in self._listeners:
-                            condition_data = self._listeners[method_name]
-                            should_trigger = False
-                            if isinstance(condition_data, tuple):
-                                _, trigger_methods = condition_data
-                                should_trigger = current_trigger in trigger_methods
-                            elif isinstance(condition_data, dict):
-                                all_methods = _extract_all_methods(condition_data)
-                                should_trigger = current_trigger in all_methods
-
-                            if should_trigger:
+                            condition_type, trigger_methods = self._listeners[
+                                method_name
+                            ]
+                            if current_trigger in trigger_methods:
                                 # Only execute if this is a cycle (method was already completed)
                                 if method_name in self._completed_methods:
                                     # For router-triggered start methods in cycles, temporarily clear resumption flag
@@ -1181,51 +1099,6 @@ class Flow(Generic[T], metaclass=FlowMeta):
                                     await self._execute_start_method(method_name)
                                     self._is_execution_resuming = was_resuming
 
-    def _evaluate_condition(
-        self, condition: str | dict, trigger_method: str, listener_name: str
-    ) -> bool:
-        """Recursively evaluate a condition (simple or nested).
-
-        Args:
-            condition: Can be a string (method name) or dict (nested condition)
-            trigger_method: The method that just completed
-            listener_name: Name of the listener being evaluated
-
-        Returns:
-            True if the condition is satisfied, False otherwise
-        """
-        if isinstance(condition, str):
-            return condition == trigger_method
-
-        if isinstance(condition, dict):
-            normalized = _normalize_condition(condition)
-            cond_type = normalized.get("type", "OR")
-            sub_conditions = normalized.get("conditions", [])
-
-            if cond_type == "OR":
-                return any(
-                    self._evaluate_condition(sub_cond, trigger_method, listener_name)
-                    for sub_cond in sub_conditions
-                )
-
-            if cond_type == "AND":
-                pending_key = f"{listener_name}:{id(condition)}"
-
-                if pending_key not in self._pending_and_listeners:
-                    all_methods = set(_extract_all_methods(condition))
-                    self._pending_and_listeners[pending_key] = all_methods
-
-                if trigger_method in self._pending_and_listeners[pending_key]:
-                    self._pending_and_listeners[pending_key].discard(trigger_method)
-
-                if not self._pending_and_listeners[pending_key]:
-                    self._pending_and_listeners.pop(pending_key, None)
-                    return True
-
-                return False
-
-        return False
-
     def _find_triggered_methods(
         self, trigger_method: str, router_only: bool
     ) -> list[str]:
@@ -1233,7 +1106,7 @@ class Flow(Generic[T], metaclass=FlowMeta):
         Finds all methods that should be triggered based on conditions.
 
         This internal method evaluates both OR and AND conditions to determine
-        which methods should be executed next in the flow. Supports nested conditions.
+        which methods should be executed next in the flow.
 
         Parameters
         ----------
@@ -1250,13 +1123,14 @@ class Flow(Generic[T], metaclass=FlowMeta):
 
         Notes
         -----
-        - Handles both OR and AND conditions, including nested combinations
+        - Handles both OR and AND conditions:
+          * OR: Triggers if any condition is met
+          * AND: Triggers only when all conditions are met
         - Maintains state for AND conditions using _pending_and_listeners
         - Separates router and normal listener evaluation
         """
         triggered = []
-
-        for listener_name, condition_data in self._listeners.items():
+        for listener_name, (condition_type, methods) in self._listeners.items():
             is_router = listener_name in self._routers
 
             if router_only != is_router:
@@ -1265,29 +1139,23 @@ class Flow(Generic[T], metaclass=FlowMeta):
             if not router_only and listener_name in self._start_methods:
                 continue
 
-            if isinstance(condition_data, tuple):
-                condition_type, methods = condition_data
-
-                if condition_type == "OR":
-                    if trigger_method in methods:
-                        triggered.append(listener_name)
-                elif condition_type == "AND":
-                    if listener_name not in self._pending_and_listeners:
-                        self._pending_and_listeners[listener_name] = set(methods)
-                    if trigger_method in self._pending_and_listeners[listener_name]:
-                        self._pending_and_listeners[listener_name].discard(
-                            trigger_method
-                        )
-
-                    if not self._pending_and_listeners[listener_name]:
-                        triggered.append(listener_name)
-                        self._pending_and_listeners.pop(listener_name, None)
-
-            elif isinstance(condition_data, dict):
-                if self._evaluate_condition(
-                    condition_data, trigger_method, listener_name
-                ):
+            if condition_type == "OR":
+                # If the trigger_method matches any in methods, run this
+                if trigger_method in methods:
                     triggered.append(listener_name)
+            elif condition_type == "AND":
+                # Initialize pending methods for this listener if not already done
+                if listener_name not in self._pending_and_listeners:
+                    self._pending_and_listeners[listener_name] = set(methods)
+                # Remove the trigger method from pending methods
+                if trigger_method in self._pending_and_listeners[listener_name]:
+                    self._pending_and_listeners[listener_name].discard(trigger_method)
+
+                if not self._pending_and_listeners[listener_name]:
+                    # All required methods have been executed
+                    triggered.append(listener_name)
+                    # Reset pending methods for this listener
+                    self._pending_and_listeners.pop(listener_name, None)
 
         return triggered
 
@@ -1350,7 +1218,7 @@ class Flow(Generic[T], metaclass=FlowMeta):
             raise
 
     def _log_flow_event(
-        self, message: str, color: PrinterColor | None = "yellow", level: str = "info"
+        self, message: str, color: str = "yellow", level: str = "info"
     ) -> None:
         """Centralized logging method for flow events.
 

src/crewai/task.py

@@ -7,7 +7,7 @@ import uuid
 import warnings
 from collections.abc import Callable
 from concurrent.futures import Future
-from copy import copy as shallow_copy
+from copy import copy
 from hashlib import md5
 from pathlib import Path
 from typing import (
@@ -672,9 +672,7 @@ Follow these guidelines:
         copied_data = {k: v for k, v in copied_data.items() if v is not None}
 
         cloned_context = (
-            self.context
-            if self.context is NOT_SPECIFIED
-            else [task_mapping[context_task.key] for context_task in self.context]
+            [task_mapping[context_task.key] for context_task in self.context]
             if isinstance(self.context, list)
             else None
         )
@@ -683,7 +681,7 @@ Follow these guidelines:
             return next((agent for agent in agents if agent.role == role), None)
 
         cloned_agent = get_agent_by_role(self.agent.role) if self.agent else None
-        cloned_tools = shallow_copy(self.tools) if self.tools else []
+        cloned_tools = copy(self.tools) if self.tools else []
 
         return self.__class__(
             **copied_data,

src/crewai/tools/base_tool.py

@@ -14,7 +14,6 @@ from pydantic import (
 from pydantic import BaseModel as PydanticBaseModel
 
 from crewai.tools.structured_tool import CrewStructuredTool
-from crewai.utilities.asyncio_utils import run_coroutine_sync
 
 
 class EnvVar(BaseModel):
@@ -91,7 +90,7 @@ class BaseTool(BaseModel, ABC):
 
         # If _run is async, we safely run it
         if asyncio.iscoroutine(result):
-            result = run_coroutine_sync(result)
+            result = asyncio.run(result)
 
         self.current_usage_count += 1
 

src/crewai/tools/structured_tool.py

@@ -8,7 +8,6 @@ from typing import TYPE_CHECKING, Any, get_type_hints
 
 from pydantic import BaseModel, Field, create_model
 
-from crewai.utilities.asyncio_utils import run_coroutine_sync
 from crewai.utilities.logger import Logger
 
 if TYPE_CHECKING:
@@ -270,12 +269,12 @@ class CrewStructuredTool:
         self._increment_usage_count()
 
         if inspect.iscoroutinefunction(self.func):
-            return run_coroutine_sync(self.func(**parsed_args, **kwargs))
+            return asyncio.run(self.func(**parsed_args, **kwargs))
 
         result = self.func(**parsed_args, **kwargs)
 
         if asyncio.iscoroutine(result):
-            return run_coroutine_sync(result)
+            return asyncio.run(result)
 
         return result
 

src/crewai/utilities/asyncio_utils.py

@@ -1,53 +0,0 @@
-"""Utilities for handling asyncio operations safely across different contexts."""
-
-import asyncio
-from collections.abc import Coroutine
-from typing import Any
-
-
-def run_coroutine_sync(coro: Coroutine) -> Any:
-    """
-    Run a coroutine synchronously, handling both cases where an event loop
-    is already running and where it's not.
-
-    This is useful when you need to run async code from sync code, but you're
-    not sure if you're already in an async context (e.g., when using asyncio.to_thread).
-
-    Args:
-        coro: The coroutine to run
-
-    Returns:
-        The result of the coroutine
-
-    Raises:
-        Any exception raised by the coroutine
-    """
-    try:
-        asyncio.get_running_loop()
-    except RuntimeError:
-        return asyncio.run(coro)
-    else:
-        import threading
-
-        result = None
-        exception = None
-
-        def run_in_new_loop():
-            nonlocal result, exception
-            try:
-                new_loop = asyncio.new_event_loop()
-                asyncio.set_event_loop(new_loop)
-                try:
-                    result = new_loop.run_until_complete(coro)
-                finally:
-                    new_loop.close()
-            except Exception as e:
-                exception = e
-
-        thread = threading.Thread(target=run_in_new_loop)
-        thread.start()
-        thread.join()
-
-        if exception:
-            raise exception
-        return result

src/crewai/utilities/printer.py

@@ -1,11 +1,6 @@
 """Utility for colored console output."""
 
-from __future__ import annotations
-
-from typing import TYPE_CHECKING, Final, Literal, NamedTuple
-
-if TYPE_CHECKING:
-    from _typeshed import SupportsWrite
+from typing import Final, Literal, NamedTuple
 
 PrinterColor = Literal[
     "purple",
@@ -59,22 +54,13 @@ class Printer:
 
     @staticmethod
     def print(
-        content: str | list[ColoredText],
-        color: PrinterColor | None = None,
-        sep: str | None = " ",
-        end: str | None = "\n",
-        file: SupportsWrite[str] | None = None,
-        flush: Literal[False] = False,
+        content: str | list[ColoredText], color: PrinterColor | None = None
     ) -> None:
         """Prints content to the console with optional color formatting.
 
         Args:
             content: Either a string or a list of ColoredText objects for multicolor output.
             color: Optional color for the text when content is a string. Ignored when content is a list.
-            sep: Separator to use between the text and color.
-            end: String appended after the last value.
-            file: A file-like object (stream); defaults to the current sys.stdout.
-            flush: Whether to forcibly flush the stream.
         """
         if isinstance(content, str):
             content = [ColoredText(content, color)]
@@ -82,9 +68,5 @@ class Printer:
             "".join(
                 f"{_COLOR_CODES[c.color] if c.color else ''}{c.text}{RESET}"
                 for c in content
-            ),
-            sep=sep,
-            end=end,
-            file=file,
-            flush=flush,
+            )
         )

tests/test_asyncio_tools.py

@@ -1,142 +0,0 @@
-"""Tests for asyncio tool execution in different contexts."""
-
-import asyncio
-from unittest.mock import patch
-
-import pytest
-
-from crewai import Agent, Crew, Task
-from crewai.tools import tool
-
-
-@tool
-async def async_test_tool(message: str) -> str:
-    """An async tool that returns a message."""
-    await asyncio.sleep(0.01)
-    return f"Processed: {message}"
-
-
-@tool
-def sync_test_tool(message: str) -> str:
-    """A sync tool that returns a message."""
-    return f"Sync: {message}"
-
-
-class TestAsyncioToolExecution:
-    """Test that tools work correctly in different asyncio contexts."""
-
-    @patch("crewai.Agent.execute_task")
-    def test_async_tool_with_asyncio_to_thread(self, mock_execute_task):
-        """Test that async tools work when crew is run with asyncio.to_thread."""
-        mock_execute_task.return_value = "Task completed"
-
-        agent = Agent(
-            role="Test Agent",
-            goal="Test async tool execution",
-            backstory="A test agent",
-        )
-
-        task = Task(
-            description="Test task",
-            expected_output="A result",
-            agent=agent,
-            tools=[async_test_tool],
-        )
-
-        crew = Crew(agents=[agent], tasks=[task], verbose=False)
-
-        async def run_with_to_thread():
-            """Run crew with asyncio.to_thread - this should not hang."""
-            result = await asyncio.to_thread(crew.kickoff)
-            return result
-
-        result = asyncio.run(run_with_to_thread())
-        assert result is not None
-
-    @patch("crewai.Agent.execute_task")
-    def test_sync_tool_with_asyncio_to_thread(self, mock_execute_task):
-        """Test that sync tools work when crew is run with asyncio.to_thread."""
-        mock_execute_task.return_value = "Task completed"
-
-        agent = Agent(
-            role="Test Agent",
-            goal="Test sync tool execution",
-            backstory="A test agent",
-        )
-
-        task = Task(
-            description="Test task",
-            expected_output="A result",
-            agent=agent,
-            tools=[sync_test_tool],
-        )
-
-        crew = Crew(agents=[agent], tasks=[task], verbose=False)
-
-        async def run_with_to_thread():
-            """Run crew with asyncio.to_thread."""
-            result = await asyncio.to_thread(crew.kickoff)
-            return result
-
-        result = asyncio.run(run_with_to_thread())
-        assert result is not None
-
-    @pytest.mark.asyncio
-    @patch("crewai.Agent.execute_task")
-    async def test_async_tool_with_kickoff_async(self, mock_execute_task):
-        """Test that async tools work with kickoff_async."""
-        mock_execute_task.return_value = "Task completed"
-
-        agent = Agent(
-            role="Test Agent",
-            goal="Test async tool execution",
-            backstory="A test agent",
-        )
-
-        task = Task(
-            description="Test task",
-            expected_output="A result",
-            agent=agent,
-            tools=[async_test_tool],
-        )
-
-        crew = Crew(agents=[agent], tasks=[task], verbose=False)
-
-        result = await crew.kickoff_async()
-        assert result is not None
-
-    def test_async_tool_direct_invocation(self):
-        """Test that async tools can be invoked directly from sync context."""
-        structured_tool = async_test_tool.to_structured_tool()
-        result = structured_tool.invoke({"message": "test"})
-        assert result == "Processed: test"
-
-    def test_async_tool_invocation_from_thread(self):
-        """Test that async tools work when invoked from a thread pool."""
-        structured_tool = async_test_tool.to_structured_tool()
-
-        def invoke_tool():
-            return structured_tool.invoke({"message": "test"})
-
-        async def run_in_thread():
-            result = await asyncio.to_thread(invoke_tool)
-            return result
-
-        result = asyncio.run(run_in_thread())
-        assert result == "Processed: test"
-
-    @pytest.mark.asyncio
-    async def test_multiple_async_tools_concurrent(self):
-        """Test multiple async tool invocations concurrently."""
-        structured_tool = async_test_tool.to_structured_tool()
-
-        async def invoke_async():
-            return await structured_tool.ainvoke({"message": "test"})
-
-        results = await asyncio.gather(
-            invoke_async(), invoke_async(), invoke_async()
-        )
-
-        assert len(results) == 3
-        for r in results:
-            assert "test" in str(r)

tests/test_flow.py

@@ -6,15 +6,15 @@ from datetime import datetime
 import pytest
 from pydantic import BaseModel
 
+from crewai.flow.flow import Flow, and_, listen, or_, router, start
 from crewai.events.event_bus import crewai_event_bus
 from crewai.events.types.flow_events import (
     FlowFinishedEvent,
-    FlowPlotEvent,
     FlowStartedEvent,
+    FlowPlotEvent,
     MethodExecutionFinishedEvent,
     MethodExecutionStartedEvent,
 )
-from crewai.flow.flow import Flow, and_, listen, or_, router, start
 
 
 def test_simple_sequential_flow():
@@ -679,11 +679,11 @@ def test_structured_flow_event_emission():
     assert isinstance(received_events[3], MethodExecutionStartedEvent)
     assert received_events[3].method_name == "send_welcome_message"
     assert received_events[3].params == {}
-    assert received_events[3].state.sent is False
+    assert getattr(received_events[3].state, "sent") is False
 
     assert isinstance(received_events[4], MethodExecutionFinishedEvent)
     assert received_events[4].method_name == "send_welcome_message"
-    assert received_events[4].state.sent is True
+    assert getattr(received_events[4].state, "sent") is True
     assert received_events[4].result == "Welcome, Anakin!"
 
     assert isinstance(received_events[5], FlowFinishedEvent)
@@ -894,75 +894,3 @@ def test_flow_name():
 
     flow = MyFlow()
     assert flow.name == "MyFlow"
-
-
-def test_nested_and_or_conditions():
-    """Test nested conditions like or_(and_(A, B), and_(C, D)).
-
-    Reproduces bug from issue #3719 where nested conditions are flattened,
-    causing premature execution.
-    """
-    execution_order = []
-
-    class NestedConditionFlow(Flow):
-        @start()
-        def method_1(self):
-            execution_order.append("method_1")
-
-        @listen(method_1)
-        def method_2(self):
-            execution_order.append("method_2")
-
-        @router(method_2)
-        def method_3(self):
-            execution_order.append("method_3")
-            # Choose b_condition path
-            return "b_condition"
-
-        @listen("b_condition")
-        def method_5(self):
-            execution_order.append("method_5")
-
-        @listen(method_5)
-        async def method_4(self):
-            execution_order.append("method_4")
-
-        @listen(or_("a_condition", "b_condition"))
-        async def method_6(self):
-            execution_order.append("method_6")
-
-        @listen(
-            or_(
-                and_("a_condition", method_6),
-                and_(method_6, method_4),
-            )
-        )
-        def method_7(self):
-            execution_order.append("method_7")
-
-        @listen(method_7)
-        async def method_8(self):
-            execution_order.append("method_8")
-
-    flow = NestedConditionFlow()
-    flow.kickoff()
-
-    # Verify execution happened
-    assert "method_1" in execution_order
-    assert "method_2" in execution_order
-    assert "method_3" in execution_order
-    assert "method_5" in execution_order
-    assert "method_4" in execution_order
-    assert "method_6" in execution_order
-    assert "method_7" in execution_order
-    assert "method_8" in execution_order
-
-    # Critical assertion: method_7 should only execute AFTER both method_6 AND method_4
-    # Since b_condition was returned, method_6 triggers on b_condition
-    # method_7 requires: (a_condition AND method_6) OR (method_6 AND method_4)
-    # The second condition (method_6 AND method_4) should be the one that triggers
-    assert execution_order.index("method_7") > execution_order.index("method_6")
-    assert execution_order.index("method_7") > execution_order.index("method_4")
-
-    # method_8 should execute after method_7
-    assert execution_order.index("method_8") > execution_order.index("method_7")

tests/test_task.py

@@ -1218,7 +1218,7 @@ def test_create_directory_false():
     assert not resolved_dir.exists()
 
     with pytest.raises(
-        RuntimeError, match=r"Directory .* does not exist and create_directory is False"
+        RuntimeError, match="Directory .* does not exist and create_directory is False"
     ):
         task._save_file("test content")
 
@@ -1635,48 +1635,3 @@ def test_task_interpolation_with_hyphens():
     assert "say hello world" in task.prompt()
 
     assert result.raw == "Hello, World!"
-
-
-def test_task_copy_with_none_context():
-    original_task = Task(
-        description="Test task",
-        expected_output="Test output",
-        context=None
-    )
-
-    new_task = original_task.copy(agents=[], task_mapping={})
-    assert original_task.context is None
-    assert new_task.context is None
-
-
-def test_task_copy_with_not_specified_context():
-    from crewai.utilities.constants import NOT_SPECIFIED
-    original_task = Task(
-        description="Test task",
-        expected_output="Test output",
-    )
-
-    new_task = original_task.copy(agents=[], task_mapping={})
-    assert original_task.context is NOT_SPECIFIED
-    assert new_task.context is NOT_SPECIFIED
-
-
-def test_task_copy_with_list_context():
-    """Test that copying a task with list context works correctly."""
-    task1 = Task(
-        description="Task 1",
-        expected_output="Output 1"
-    )
-    task2 = Task(
-        description="Task 2",
-        expected_output="Output 2",
-        context=[task1]
-    )
-
-    task_mapping = {task1.key: task1}
-
-    copied_task2 = task2.copy(agents=[], task_mapping=task_mapping)
-
-    assert isinstance(copied_task2.context, list)
-    assert len(copied_task2.context) == 1
-    assert copied_task2.context[0] is task1