Fix type-checker: use PrinterColor type for color parameter

Co-Authored-By: João <joao@crewai.com>

.github/security.md

@@ -1,27 +1,50 @@
-## CrewAI Security Vulnerability Reporting Policy
+## CrewAI Security Policy
 
-CrewAI prioritizes the security of our software products, services, and GitHub repositories. To promptly address vulnerabilities, follow these steps for reporting security issues:
+We are committed to protecting the confidentiality, integrity, and availability of the CrewAI ecosystem. This policy explains how to report potential vulnerabilities and what you can expect from us when you do.
 
-### Reporting Process
-Do **not** report vulnerabilities via public GitHub issues.
+### Scope
 
-Email all vulnerability reports directly to:  
-**security@crewai.com**
+We welcome reports for vulnerabilities that could impact:
 
-### Required Information
-To help us quickly validate and remediate the issue, your report must include:
+- CrewAI-maintained source code and repositories
+- CrewAI-operated infrastructure and services
+- Official CrewAI releases, packages, and distributions
 
-- **Vulnerability Type:** Clearly state the vulnerability type (e.g., SQL injection, XSS, privilege escalation).
-- **Affected Source Code:** Provide full file paths and direct URLs (branch, tag, or commit).
-- **Reproduction Steps:** Include detailed, step-by-step instructions. Screenshots are recommended.
-- **Special Configuration:** Document any special settings or configurations required to reproduce.
-- **Proof-of-Concept (PoC):** Provide exploit or PoC code (if available).
-- **Impact Assessment:** Clearly explain the severity and potential exploitation scenarios.
+Issues affecting clearly unaffiliated third-party services or user-generated content are out of scope, unless you can demonstrate a direct impact on CrewAI systems or customers.
 
-### Our Response
-- We will acknowledge receipt of your report promptly via your provided email.
-- Confirmed vulnerabilities will receive priority remediation based on severity.
-- Patches will be released as swiftly as possible following verification.
+### How to Report
 
-### Reward Notice
-Currently, we do not offer a bug bounty program. Rewards, if issued, are discretionary.
+- **Please do not** disclose vulnerabilities via public GitHub issues, pull requests, or social media.
+- Email detailed reports to **security@crewai.com** with the subject line `Security Report`.
+- If you need to share large files or sensitive artifacts, mention it in your email and we will coordinate a secure transfer method.
+
+### What to Include
+
+Providing comprehensive information enables us to validate the issue quickly:
+
+- **Vulnerability overview** — a concise description and classification (e.g., RCE, privilege escalation)
+- **Affected components** — repository, branch, tag, or deployed service along with relevant file paths or endpoints
+- **Reproduction steps** — detailed, step-by-step instructions; include logs, screenshots, or screen recordings when helpful
+- **Proof-of-concept** — exploit details or code that demonstrates the impact (if available)
+- **Impact analysis** — severity assessment, potential exploitation scenarios, and any prerequisites or special configurations
+
+### Our Commitment
+
+- **Acknowledgement:** We aim to acknowledge your report within two business days.
+- **Communication:** We will keep you informed about triage results, remediation progress, and planned release timelines.
+- **Resolution:** Confirmed vulnerabilities will be prioritized based on severity and fixed as quickly as possible.
+- **Recognition:** We currently do not run a bug bounty program; any rewards or recognition are issued at CrewAI's discretion.
+
+### Coordinated Disclosure
+
+We ask that you allow us a reasonable window to investigate and remediate confirmed issues before any public disclosure. We will coordinate publication timelines with you whenever possible.
+
+### Safe Harbor
+
+We will not pursue or support legal action against individuals who, in good faith:
+
+- Follow this policy and refrain from violating any applicable laws
+- Avoid privacy violations, data destruction, or service disruption
+- Limit testing to systems in scope and respect rate limits and terms of service
+
+If you are unsure whether your testing is covered, please contact us at **security@crewai.com** before proceeding.

src/crewai/__init__.py

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

src/crewai/cli/authentication/utils.py

@@ -30,6 +30,7 @@ def validate_jwt_token(
             algorithms=["RS256"],
             audience=audience,
             issuer=issuer,
+            leeway=10.0,
             options={
                 "verify_signature": True,
                 "verify_exp": True,

src/crewai/cli/run_crew.py

@@ -1,10 +1,11 @@
+import os
 import subprocess
 from enum import Enum
 
 import click
 from packaging import version
 
-from crewai.cli.utils import read_toml
+from crewai.cli.utils import build_env_with_tool_repository_credentials, read_toml
 from crewai.cli.version import get_crewai_version
 
 
@@ -55,8 +56,22 @@ def execute_command(crew_type: CrewType) -> None:
     """
     command = ["uv", "run", "kickoff" if crew_type == CrewType.FLOW else "run_crew"]
 
+    env = os.environ.copy()
     try:
-        subprocess.run(command, capture_output=False, text=True, check=True)  # noqa: S603
+        pyproject_data = read_toml()
+        sources = pyproject_data.get("tool", {}).get("uv", {}).get("sources", {})
+
+        for source_config in sources.values():
+            if isinstance(source_config, dict):
+                index = source_config.get("index")
+                if index:
+                    index_env = build_env_with_tool_repository_credentials(index)
+                    env.update(index_env)
+    except Exception:  # noqa: S110
+        pass
+
+    try:
+        subprocess.run(command, capture_output=False, text=True, check=True, env=env)  # noqa: S603
 
     except subprocess.CalledProcessError as e:
         handle_error(e, crew_type)

src/crewai/cli/templates/crew/pyproject.toml

@@ -5,7 +5,7 @@ description = "{{name}} using crewAI"
 authors = [{ name = "Your Name", email = "you@example.com" }]
 requires-python = ">=3.10,<3.14"
 dependencies = [
-    "crewai[tools]>=0.203.0,<1.0.0"
+    "crewai[tools]>=0.203.1,<1.0.0"
 ]
 
 [project.scripts]

src/crewai/cli/templates/flow/pyproject.toml

@@ -5,7 +5,7 @@ description = "{{name}} using crewAI"
 authors = [{ name = "Your Name", email = "you@example.com" }]
 requires-python = ">=3.10,<3.14"
 dependencies = [
-    "crewai[tools]>=0.203.0,<1.0.0",
+    "crewai[tools]>=0.203.1,<1.0.0",
 ]
 
 [project.scripts]

src/crewai/cli/templates/tool/pyproject.toml

@@ -5,7 +5,7 @@ description = "Power up your crews with {{folder_name}}"
 readme = "README.md"
 requires-python = ">=3.10,<3.14"
 dependencies = [
-    "crewai[tools]>=0.203.0"
+    "crewai[tools]>=0.203.1"
 ]
 
 [tool.crewai]

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

@@ -358,7 +358,8 @@ 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):
+            except (EOFError, KeyboardInterrupt, OSError, LookupError):
+                # Handle all input-related errors silently
                 result[0] = False
 
         input_thread = threading.Thread(target=get_input, daemon=True)
@@ -371,6 +372,7 @@ 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
+from crewai.utilities.printer import Printer, PrinterColor
 
 logger = logging.getLogger(__name__)
 
@@ -459,6 +459,7 @@ class Flow(Generic[T], metaclass=FlowMeta):
         self._methods: dict[str, Callable] = {}
         self._method_execution_counts: dict[str, int] = {}
         self._pending_and_listeners: dict[str, set[str]] = {}
+        self._triggered_or_listeners: set[str] = set()  # Track OR listeners that have already triggered
         self._method_outputs: list[Any] = []  # list to store all method outputs
         self._completed_methods: set[str] = set()  # Track completed methods for reload
         self._persistence: FlowPersistence | None = persistence
@@ -822,6 +823,7 @@ class Flow(Generic[T], metaclass=FlowMeta):
                 # Clear completed methods and outputs for a fresh start
                 self._completed_methods.clear()
                 self._method_outputs.clear()
+                self._triggered_or_listeners.clear()
             else:
                 # We're restoring from persistence, set the flag
                 self._is_execution_resuming = True
@@ -922,6 +924,7 @@ class Flow(Generic[T], metaclass=FlowMeta):
                 return
             # For cyclic flows, clear from completed to allow re-execution
             self._completed_methods.discard(start_method_name)
+            self._triggered_or_listeners.clear()
 
         method = self._methods[start_method_name]
         enhanced_method = self._inject_trigger_payload_for_start_method(method)
@@ -1086,7 +1089,7 @@ 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_type, trigger_methods = self._listeners[
+                            _, trigger_methods = self._listeners[
                                 method_name
                             ]
                             if current_trigger in trigger_methods:
@@ -1124,9 +1127,10 @@ class Flow(Generic[T], metaclass=FlowMeta):
         Notes
         -----
         - Handles both OR and AND conditions:
-          * OR: Triggers if any condition is met
+          * OR: Triggers if any condition is met (but only once per listener)
           * AND: Triggers only when all conditions are met
         - Maintains state for AND conditions using _pending_and_listeners
+        - Tracks OR listeners to prevent multiple triggers from the same condition
         - Separates router and normal listener evaluation
         """
         triggered = []
@@ -1140,9 +1144,15 @@ class Flow(Generic[T], metaclass=FlowMeta):
                 continue
 
             if condition_type == "OR":
+                # Check if this OR listener has already been triggered
+                if listener_name in self._triggered_or_listeners:
+                    # Skip this listener as it has already been triggered by another method in the OR condition
+                    continue
+
                 # If the trigger_method matches any in methods, run this
                 if trigger_method in methods:
                     triggered.append(listener_name)
+                    self._triggered_or_listeners.add(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:
@@ -1195,6 +1205,7 @@ class Flow(Generic[T], metaclass=FlowMeta):
                 return
             # For cyclic flows, clear from completed to allow re-execution
             self._completed_methods.discard(listener_name)
+            self._triggered_or_listeners.discard(listener_name)
 
         try:
             method = self._methods[listener_name]
@@ -1218,7 +1229,7 @@ class Flow(Generic[T], metaclass=FlowMeta):
             raise
 
     def _log_flow_event(
-        self, message: str, color: str = "yellow", level: str = "info"
+        self, message: str, color: PrinterColor = "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
+from copy import copy as shallow_copy
 from hashlib import md5
 from pathlib import Path
 from typing import (
@@ -672,7 +672,9 @@ Follow these guidelines:
         copied_data = {k: v for k, v in copied_data.items() if v is not None}
 
         cloned_context = (
-            [task_mapping[context_task.key] for context_task in self.context]
+            self.context
+            if self.context is NOT_SPECIFIED
+            else [task_mapping[context_task.key] for context_task in self.context]
             if isinstance(self.context, list)
             else None
         )
@@ -681,7 +683,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 = copy(self.tools) if self.tools else []
+        cloned_tools = shallow_copy(self.tools) if self.tools else []
 
         return self.__class__(
             **copied_data,

tests/cli/authentication/test_utils.py

@@ -1,7 +1,7 @@
-import jwt
 import unittest
 from unittest.mock import MagicMock, patch
 
+import jwt
 
 from crewai.cli.authentication.utils import validate_jwt_token
 
@@ -17,19 +17,22 @@ class TestUtils(unittest.TestCase):
             key="mock_signing_key"
         )
 
+        jwt_token = "aaaaa.bbbbbb.cccccc"  # noqa: S105
+
         decoded_token = validate_jwt_token(
-            jwt_token="aaaaa.bbbbbb.cccccc",
+            jwt_token=jwt_token,
             jwks_url="https://mock_jwks_url",
             issuer="https://mock_issuer",
             audience="app_id_xxxx",
         )
 
         mock_jwt.decode.assert_called_with(
-            "aaaaa.bbbbbb.cccccc",
+            jwt_token,
             "mock_signing_key",
             algorithms=["RS256"],
             audience="app_id_xxxx",
             issuer="https://mock_issuer",
+            leeway=10.0,
             options={
                 "verify_signature": True,
                 "verify_exp": True,
@@ -43,9 +46,9 @@ class TestUtils(unittest.TestCase):
 
     def test_validate_jwt_token_expired(self, mock_jwt, mock_pyjwkclient):
         mock_jwt.decode.side_effect = jwt.ExpiredSignatureError
-        with self.assertRaises(Exception):
+        with self.assertRaises(Exception):  # noqa: B017
             validate_jwt_token(
-                jwt_token="aaaaa.bbbbbb.cccccc",
+                jwt_token="aaaaa.bbbbbb.cccccc",  # noqa: S106
                 jwks_url="https://mock_jwks_url",
                 issuer="https://mock_issuer",
                 audience="app_id_xxxx",
@@ -53,9 +56,9 @@ class TestUtils(unittest.TestCase):
 
     def test_validate_jwt_token_invalid_audience(self, mock_jwt, mock_pyjwkclient):
         mock_jwt.decode.side_effect = jwt.InvalidAudienceError
-        with self.assertRaises(Exception):
+        with self.assertRaises(Exception):  # noqa: B017
             validate_jwt_token(
-                jwt_token="aaaaa.bbbbbb.cccccc",
+                jwt_token="aaaaa.bbbbbb.cccccc",  # noqa: S106
                 jwks_url="https://mock_jwks_url",
                 issuer="https://mock_issuer",
                 audience="app_id_xxxx",
@@ -63,9 +66,9 @@ class TestUtils(unittest.TestCase):
 
     def test_validate_jwt_token_invalid_issuer(self, mock_jwt, mock_pyjwkclient):
         mock_jwt.decode.side_effect = jwt.InvalidIssuerError
-        with self.assertRaises(Exception):
+        with self.assertRaises(Exception):  # noqa: B017
             validate_jwt_token(
-                jwt_token="aaaaa.bbbbbb.cccccc",
+                jwt_token="aaaaa.bbbbbb.cccccc",  # noqa: S106
                 jwks_url="https://mock_jwks_url",
                 issuer="https://mock_issuer",
                 audience="app_id_xxxx",
@@ -75,9 +78,9 @@ class TestUtils(unittest.TestCase):
         self, mock_jwt, mock_pyjwkclient
     ):
         mock_jwt.decode.side_effect = jwt.MissingRequiredClaimError
-        with self.assertRaises(Exception):
+        with self.assertRaises(Exception):  # noqa: B017
             validate_jwt_token(
-                jwt_token="aaaaa.bbbbbb.cccccc",
+                jwt_token="aaaaa.bbbbbb.cccccc",  # noqa: S106
                 jwks_url="https://mock_jwks_url",
                 issuer="https://mock_issuer",
                 audience="app_id_xxxx",
@@ -85,9 +88,9 @@ class TestUtils(unittest.TestCase):
 
     def test_validate_jwt_token_jwks_error(self, mock_jwt, mock_pyjwkclient):
         mock_jwt.decode.side_effect = jwt.exceptions.PyJWKClientError
-        with self.assertRaises(Exception):
+        with self.assertRaises(Exception):  # noqa: B017
             validate_jwt_token(
-                jwt_token="aaaaa.bbbbbb.cccccc",
+                jwt_token="aaaaa.bbbbbb.cccccc",  # noqa: S106
                 jwks_url="https://mock_jwks_url",
                 issuer="https://mock_issuer",
                 audience="app_id_xxxx",
@@ -95,9 +98,9 @@ class TestUtils(unittest.TestCase):
 
     def test_validate_jwt_token_invalid_token(self, mock_jwt, mock_pyjwkclient):
         mock_jwt.decode.side_effect = jwt.InvalidTokenError
-        with self.assertRaises(Exception):
+        with self.assertRaises(Exception):  # noqa: B017
             validate_jwt_token(
-                jwt_token="aaaaa.bbbbbb.cccccc",
+                jwt_token="aaaaa.bbbbbb.cccccc",  # noqa: S106
                 jwks_url="https://mock_jwks_url",
                 issuer="https://mock_issuer",
                 audience="app_id_xxxx",

tests/test_nested_or_conditions.py

@@ -0,0 +1,296 @@
+"""Test nested or_() conditions in Flow execution (Issue #3719)."""
+
+
+from crewai.flow.flow import Flow, listen, or_, start
+
+
+def test_nested_or_condition_triggers_once():
+    """Test that nested or_() conditions only trigger listeners once.
+
+    This test addresses issue #3719 where nested or_() conditions would
+    cause listeners to execute multiple times instead of once.
+
+    Setup:
+        method_5 listens to or_(method_1, or_(method_2, method_3))
+        method_7 listens to or_(method_5, method_6)
+
+    Expected behavior:
+        - method_5 should execute exactly once (triggered by first matching condition)
+        - method_7 should execute exactly once (triggered by first matching condition)
+
+    Bug behavior (before fix):
+        - method_5 executed 3 times (once for method_1, method_2, and method_3)
+        - method_7 executed 4 times (once for each method_5 execution + method_6)
+    """
+    execution_order = []
+
+    class NestedOrFlow(Flow):
+        @start()
+        def method_1(self):
+            execution_order.append("method_1")
+            return "method_1_done"
+
+        @listen("method_1")
+        def method_2(self):
+            execution_order.append("method_2")
+            return "method_2_done"
+
+        @listen("method_1")
+        def method_3(self):
+            execution_order.append("method_3")
+            return "method_3_done"
+
+        @listen(or_("method_1", or_("method_2", "method_3")))
+        def method_5(self):
+            execution_order.append("method_5")
+            return "method_5_done"
+
+        @listen("method_1")
+        def method_6(self):
+            execution_order.append("method_6")
+            return "method_6_done"
+
+        @listen(or_("method_5", "method_6"))
+        def method_7(self):
+            execution_order.append("method_7")
+            return "method_7_done"
+
+    flow = NestedOrFlow()
+    flow.kickoff()
+
+    assert execution_order.count("method_5") == 1, (
+        f"method_5 should execute exactly once, but executed {execution_order.count('method_5')} times"
+    )
+    assert execution_order.count("method_7") == 1, (
+        f"method_7 should execute exactly once, but executed {execution_order.count('method_7')} times"
+    )
+
+    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_6" in execution_order
+    assert "method_7" in execution_order
+
+
+def test_simple_or_condition_triggers_once():
+    """Test that simple or_() conditions only trigger once.
+
+    Even without nesting, an OR condition should only trigger a listener once,
+    not multiple times for each method in the OR list.
+    """
+    execution_order = []
+
+    class SimpleOrFlow(Flow):
+        @start()
+        def method_a(self):
+            execution_order.append("method_a")
+
+        @listen("method_a")
+        def method_b(self):
+            execution_order.append("method_b")
+
+        @listen("method_a")
+        def method_c(self):
+            execution_order.append("method_c")
+
+        @listen(or_("method_b", "method_c"))
+        def method_d(self):
+            execution_order.append("method_d")
+
+    flow = SimpleOrFlow()
+    flow.kickoff()
+
+    assert execution_order.count("method_d") == 1, (
+        f"method_d should execute exactly once, but executed {execution_order.count('method_d')} times"
+    )
+
+
+def test_or_condition_with_three_methods():
+    """Test OR condition with three methods triggers only once."""
+    execution_order = []
+
+    class ThreeMethodOrFlow(Flow):
+        @start()
+        def method_1(self):
+            execution_order.append("method_1")
+
+        @listen("method_1")
+        def method_2(self):
+            execution_order.append("method_2")
+
+        @listen("method_1")
+        def method_3(self):
+            execution_order.append("method_3")
+
+        @listen("method_1")
+        def method_4(self):
+            execution_order.append("method_4")
+
+        @listen(or_("method_2", "method_3", "method_4"))
+        def method_5(self):
+            execution_order.append("method_5")
+
+    flow = ThreeMethodOrFlow()
+    flow.kickoff()
+
+    assert execution_order.count("method_5") == 1, (
+        f"method_5 should execute exactly once, but executed {execution_order.count('method_5')} times"
+    )
+
+
+def test_multiple_or_listeners_independent():
+    """Test that multiple OR listeners are independent of each other."""
+    execution_order = []
+
+    class MultipleOrFlow(Flow):
+        @start()
+        def method_1(self):
+            execution_order.append("method_1")
+
+        @listen("method_1")
+        def method_2(self):
+            execution_order.append("method_2")
+
+        @listen("method_1")
+        def method_3(self):
+            execution_order.append("method_3")
+
+        @listen(or_("method_2", "method_3"))
+        def method_a(self):
+            execution_order.append("method_a")
+
+        @listen(or_("method_2", "method_3"))
+        def method_b(self):
+            execution_order.append("method_b")
+
+    flow = MultipleOrFlow()
+    flow.kickoff()
+
+    assert execution_order.count("method_a") == 1
+    assert execution_order.count("method_b") == 1
+
+
+def test_deeply_nested_or_conditions():
+    """Test deeply nested or_() conditions."""
+    execution_order = []
+
+    class DeeplyNestedOrFlow(Flow):
+        @start()
+        def start_method(self):
+            execution_order.append("start_method")
+
+        @listen("start_method")
+        def method_a(self):
+            execution_order.append("method_a")
+
+        @listen("start_method")
+        def method_b(self):
+            execution_order.append("method_b")
+
+        @listen("start_method")
+        def method_c(self):
+            execution_order.append("method_c")
+
+        @listen("start_method")
+        def method_d(self):
+            execution_order.append("method_d")
+
+        @listen(or_(or_("method_a", "method_b"), or_("method_c", "method_d")))
+        def final_method(self):
+            execution_order.append("final_method")
+
+    flow = DeeplyNestedOrFlow()
+    flow.kickoff()
+
+    assert execution_order.count("final_method") == 1, (
+        f"final_method should execute exactly once, but executed {execution_order.count('final_method')} times"
+    )
+
+
+def test_or_condition_execution_order():
+    """Test that OR listener executes after first matching condition.
+
+    The listener should trigger as soon as any one of the OR conditions is met,
+    not wait for all of them.
+    """
+    execution_order = []
+
+    class ExecutionOrderFlow(Flow):
+        @start()
+        def method_1(self):
+            execution_order.append("method_1")
+
+        @listen("method_1")
+        def method_2(self):
+            execution_order.append("method_2")
+
+        @listen("method_1")
+        def method_3(self):
+            execution_order.append("method_3")
+
+        @listen(or_("method_2", "method_3"))
+        def method_4(self):
+            execution_order.append("method_4")
+
+    flow = ExecutionOrderFlow()
+    flow.kickoff()
+
+    method_4_index = execution_order.index("method_4")
+
+    assert "method_2" in execution_order[:method_4_index] or "method_3" in execution_order[:method_4_index], (
+        "method_4 should execute after at least one of method_2 or method_3"
+    )
+
+
+def test_or_condition_with_single_method():
+    """Test OR condition with a single method (edge case)."""
+    execution_order = []
+
+    class SingleMethodOrFlow(Flow):
+        @start()
+        def method_1(self):
+            execution_order.append("method_1")
+
+        @listen(or_("method_1"))
+        def method_2(self):
+            execution_order.append("method_2")
+
+    flow = SingleMethodOrFlow()
+    flow.kickoff()
+
+    assert execution_order == ["method_1", "method_2"]
+    assert execution_order.count("method_2") == 1
+
+
+def test_cyclic_flow_with_or_condition():
+    """Test that OR conditions work correctly in cyclic flows.
+
+    Within a single flow execution, an OR listener should only trigger once
+    even if multiple methods in the OR condition complete.
+    """
+    execution_order = []
+
+    class CyclicOrFlow(Flow):
+        @start()
+        def step_1(self):
+            execution_order.append("step_1")
+
+        @listen("step_1")
+        def step_2(self):
+            execution_order.append("step_2")
+
+        @listen("step_1")
+        def step_3(self):
+            execution_order.append("step_3")
+
+        @listen(or_("step_2", "step_3"))
+        def step_4(self):
+            execution_order.append("step_4")
+
+    flow = CyclicOrFlow()
+    flow.kickoff()
+
+    assert execution_order.count("step_4") == 1, (
+        f"step_4 should execute once (not once for each OR condition), but executed {execution_order.count('step_4')} times"
+    )

tests/test_task.py

@@ -1218,7 +1218,7 @@ def test_create_directory_false():
     assert not resolved_dir.exists()
 
     with pytest.raises(
-        RuntimeError, match="Directory .* does not exist and create_directory is False"
+        RuntimeError, match=r"Directory .* does not exist and create_directory is False"
     ):
         task._save_file("test content")
 
@@ -1635,3 +1635,48 @@ 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