fix: patch event-bus singleton in LLM emit unit tests

Class-level CrewAIEventsBus.emit patches are unreliable under
pytest --import-mode=importlib / xdist. Patch the singleton instance
instead so streaming finish-reason and usage-event tests observe emits.

Co-authored-by: Rip&Tear <theCyberTech@users.noreply.github.com>

lib/crewai/src/crewai/llms/providers/azure/completion.py

@@ -183,13 +183,24 @@ class AzureCompletion(BaseLLM):
             pass
         return self
 
+    @staticmethod
+    def _openai_completion_class() -> Any:
+        """Return the OpenAICompletion class used for Responses API delegation.
+
+        Isolated so tests can patch this lookup reliably under pytest-xdist
+        instead of racing the dynamic import inside ``_init_responses_delegate``.
+        """
+        from crewai.llms.providers.openai.completion import OpenAICompletion
+
+        return OpenAICompletion
+
     def _init_responses_delegate(self) -> None:
         """Create an OpenAICompletion delegate for the Azure OpenAI Responses API.
 
         The Azure OpenAI Responses API uses the standard OpenAI Python SDK
         with a base_url pointing to the Azure resource's /openai/v1/ endpoint.
         """
-        from crewai.llms.providers.openai.completion import OpenAICompletion
+        openai_completion_cls = self._openai_completion_class()
 
         base_url = self._get_responses_base_url()
 
@@ -239,7 +250,7 @@ class AzureCompletion(BaseLLM):
         if self.additional_params:
             delegate_kwargs["additional_params"] = self.additional_params
 
-        self._responses_delegate = OpenAICompletion(**delegate_kwargs)
+        self._responses_delegate = openai_completion_cls(**delegate_kwargs)
 
     def _get_responses_base_url(self) -> str:
         """Construct the base URL for the Azure OpenAI Responses API.

lib/crewai/src/crewai/task.py

@@ -10,7 +10,7 @@ from hashlib import md5
 import inspect
 import json
 import logging
-from pathlib import Path
+from pathlib import Path, PurePosixPath, PureWindowsPath
 import threading
 from typing import (
     Annotated,
@@ -517,6 +517,29 @@ class Task(BaseModel):
         if value is None:
             return None
 
+        if "{" in value or "}" in value:
+            template_vars = [part.split("}")[0] for part in value.split("{")[1:]]
+            for var in template_vars:
+                if not var.isidentifier():
+                    raise ValueError(f"Invalid template variable name: {var}")
+            # Literal portions are still checked here; the fully interpolated
+            # path is re-validated at runtime (see
+            # interpolate_inputs_and_add_conversation_history) because template
+            # variables may be filled from untrusted kickoff inputs.
+            cls._sanitize_output_file_path(value)
+            return value
+
+        return cls._sanitize_output_file_path(value)
+
+    @staticmethod
+    def _sanitize_output_file_path(value: str) -> str:
+        """Enforce path-safety on an ``output_file`` value.
+
+        Shared by the field validator's literal and template branches. Rejects
+        traversal sequences, shell expansion, and shell metacharacters, and
+        strips a leading ``/`` so a literal path stays relative to the working
+        directory.
+        """
         if ".." in value:
             raise ValueError(
                 "Path traversal attempts are not allowed in output_file paths"
@@ -532,17 +555,109 @@ class Task(BaseModel):
                 "Shell special characters are not allowed in output_file paths"
             )
 
-        if "{" in value or "}" in value:
-            template_vars = [part.split("}")[0] for part in value.split("{")[1:]]
-            for var in template_vars:
-                if not var.isidentifier():
-                    raise ValueError(f"Invalid template variable name: {var}")
-            return value
-
         if value.startswith("/"):
             return value[1:]
         return value
 
+    @staticmethod
+    def _is_unsafe_absolute_output_path(value: str) -> bool:
+        """Return True if ``value`` is an absolute or drive/root-relative path.
+
+        Includes Windows drive-qualified relative paths (``C:foo``) and
+        root-relative paths (``\\Windows\\...``) that ``is_absolute()`` misses
+        but that still escape a relative working directory on Windows.
+        """
+        windows_path = PureWindowsPath(value)
+        return bool(
+            PurePosixPath(value).is_absolute()
+            or windows_path.is_absolute()
+            or windows_path.drive
+            or windows_path.root
+        )
+
+    def _validate_output_file_input_values(
+        self, inputs: dict[str, str | int | float | dict[str, Any] | list[Any]]
+    ) -> None:
+        """Reject untrusted input values that would escape the output path.
+
+        Only the variables that actually appear in the ``output_file`` template
+        are checked. The developer-authored template is trusted (it may contain
+        an absolute base directory), but a value substituted into it must not
+        introduce path traversal (``..``), an absolute path, a home/variable
+        expansion (``~``/``$``), or shell metacharacters that would redirect the
+        write outside the intended location.
+        """
+        if not self._original_output_file:
+            return
+
+        template_vars = [
+            part.split("}")[0] for part in self._original_output_file.split("{")[1:]
+        ]
+        for var in template_vars:
+            if var not in inputs:
+                continue
+            value = str(inputs[var])
+            if ".." in value:
+                raise ValueError(
+                    f"Invalid value for output_file variable '{var}': Path "
+                    "traversal sequences ('..') are not allowed"
+                )
+            if value.startswith(("~", "$")):
+                raise ValueError(
+                    f"Invalid value for output_file variable '{var}': Shell "
+                    "expansion characters are not allowed"
+                )
+            if any(char in value for char in ["|", ">", "<", "&", ";"]):
+                raise ValueError(
+                    f"Invalid value for output_file variable '{var}': Shell "
+                    "special characters are not allowed"
+                )
+            if self._is_unsafe_absolute_output_path(value):
+                raise ValueError(
+                    f"Invalid value for output_file variable '{var}': Absolute "
+                    "paths are not allowed"
+                )
+
+    def _validate_interpolated_output_file(self, interpolated: str) -> None:
+        """Reject a fully interpolated path that escapes the trusted template.
+
+        Per-value checks miss hazards formed by concatenating adjacent
+        placeholders (for example ``{a}{b}`` with ``a='.'`` and ``b='.'``).
+        The developer-authored template may include an absolute base directory;
+        that remains allowed when the interpolated result stays absolute for the
+        same reason. A relative template must not become absolute, and no
+        interpolated path may introduce traversal or shell metacharacters.
+        """
+        template = self._original_output_file
+        if not template:
+            return
+
+        if ".." in interpolated:
+            raise ValueError(
+                "Path traversal attempts are not allowed in output_file paths"
+            )
+        if interpolated.startswith(("~", "$")):
+            raise ValueError(
+                "Shell expansion characters are not allowed in output_file paths"
+            )
+        if any(char in interpolated for char in ["|", ">", "<", "&", ";"]):
+            raise ValueError(
+                "Shell special characters are not allowed in output_file paths"
+            )
+
+        dummy_filled = template
+        for var in (part.split("}")[0] for part in template.split("{")[1:]):
+            dummy_filled = dummy_filled.replace("{" + var + "}", "_x_")
+
+        template_is_absolute = self._is_unsafe_absolute_output_path(dummy_filled)
+        if (
+            self._is_unsafe_absolute_output_path(interpolated)
+            and not template_is_absolute
+        ):
+            raise ValueError(
+                "Absolute paths are not allowed in interpolated output_file paths"
+            )
+
     @model_validator(mode="after")
     def set_attributes_based_on_config(self) -> Task:
         """Set attributes based on the agent configuration."""
@@ -1101,12 +1216,23 @@ Follow these guidelines:
             raise ValueError(f"Error interpolating expected_output: {e!s}") from e
 
         if self.output_file is not None:
+            # Values interpolated into the output path may come from untrusted
+            # kickoff inputs. The developer-authored template (including any
+            # absolute base directory) is trusted, but an injected value must
+            # not introduce path traversal, an absolute path, or shell
+            # expansion that would escape the intended location. Per-value
+            # checks run first; the fully interpolated path is then checked so
+            # concatenated placeholders cannot form a hazard the individual
+            # values alone would miss.
+            self._validate_output_file_input_values(inputs)
             try:
-                self.output_file = interpolate_only(
+                interpolated_output_file = interpolate_only(
                     input_string=self._original_output_file, inputs=inputs
                 )
             except (KeyError, ValueError) as e:
                 raise ValueError(f"Error interpolating output_file path: {e!s}") from e
+            self._validate_interpolated_output_file(interpolated_output_file)
+            self.output_file = interpolated_output_file
 
         if inputs.get("crew_chat_messages"):
             conversation_instruction = I18N_DEFAULT.slice(

lib/crewai/tests/events/test_llm_usage_event.py

@@ -4,7 +4,7 @@ from unittest.mock import patch
 import pytest
 from pydantic import BaseModel
 
-from crewai.events.event_bus import CrewAIEventsBus
+from crewai.events.event_bus import crewai_event_bus
 from crewai.events.types.llm_events import LLMCallCompletedEvent, LLMCallType
 from crewai.llm import LLM
 from crewai.llms.base_llm import BaseLLM
@@ -203,7 +203,9 @@ class _StubLLM(BaseLLM):
 class TestEmitCallCompletedEventPassesUsage:
     @pytest.fixture
     def mock_emit(self):
-        with patch.object(CrewAIEventsBus, "emit") as mock:
+        # Patch the singleton instance; class-level patches are unreliable
+        # under pytest ``--import-mode=importlib`` / xdist.
+        with patch.object(crewai_event_bus, "emit") as mock:
             yield mock
 
     @pytest.fixture

lib/crewai/tests/llms/azure/test_azure_responses.py

@@ -29,9 +29,12 @@ def azure_env():
 def mock_openai_completion():
     """Mock OpenAICompletion to avoid real client creation.
 
-    Patches at the source module so that the dynamic import inside
-    _init_responses_delegate picks up the mock.
+    Patches ``AzureCompletion._openai_completion_class`` so the Responses
+    delegate lookup is deterministic under pytest-xdist (patching the
+    dynamic import target alone can miss under parallel workers).
     """
+    from crewai.llms.providers.azure.completion import AzureCompletion
+
     instance = MagicMock()
     instance.call = MagicMock(return_value="responses-result")
     instance.acall = AsyncMock(return_value="async-responses-result")
@@ -41,9 +44,10 @@ def mock_openai_completion():
     instance.reset_reasoning_chain = MagicMock()
     mock_cls = MagicMock(return_value=instance)
 
-    with patch(
-        "crewai.llms.providers.openai.completion.OpenAICompletion",
-        mock_cls,
+    with patch.object(
+        AzureCompletion,
+        "_openai_completion_class",
+        return_value=mock_cls,
     ):
         yield mock_cls, instance
 

lib/crewai/tests/test_llm_streaming_finish_reason.py

@@ -11,22 +11,33 @@ from unittest.mock import patch
 
 import pytest
 
-from crewai.events.event_bus import CrewAIEventsBus
+from crewai.events.event_bus import crewai_event_bus
 from crewai.events.types.llm_events import LLMCallCompletedEvent
 from crewai.llm import LLM
 
 
 @pytest.fixture
 def mock_emit():
-    with patch.object(CrewAIEventsBus, "emit") as mock:
+    # Patch the singleton instance (not the class). Class-level patches are
+    # unreliable under pytest ``--import-mode=importlib`` / xdist because the
+    # test and ``crewai.llm`` can observe different class objects.
+    with patch.object(crewai_event_bus, "emit") as mock:
         yield mock
 
 
+def _event_from_call(call) -> object | None:
+    if "event" in call.kwargs:
+        return call.kwargs["event"]
+    if len(call.args) >= 2:
+        return call.args[1]
+    return None
+
+
 def _completed_event(mock_emit) -> LLMCallCompletedEvent:
     matches = [
-        call.kwargs["event"]
+        event
         for call in mock_emit.call_args_list
-        if isinstance(call.kwargs.get("event"), LLMCallCompletedEvent)
+        if isinstance((event := _event_from_call(call)), LLMCallCompletedEvent)
     ]
     assert matches, "expected an LLMCallCompletedEvent to be emitted"
     assert len(matches) == 1, f"expected one completed event, got {len(matches)}"

lib/crewai/tests/test_task.py

@@ -932,6 +932,53 @@ def test_interpolate_inputs(tmp_path):
     assert task.output_file == str(tmp_path / "ML" / "output_2025.txt")
 
 
+@pytest.mark.parametrize(
+    ("template", "malicious_inputs", "expected_error"),
+    [
+        ("reports/{name}.md", {"name": "../../../../tmp/pwn"}, "Path traversal"),
+        ("{p}", {"p": "/tmp/abs_pwn"}, "Absolute paths"),
+        ("{p}", {"p": "~/.bashrc"}, "Shell expansion"),
+        ("{p}", {"p": "x;rm -rf /"}, "Shell special characters"),
+        ("{p}", {"p": r"C:\Windows\evil"}, "Absolute paths"),
+        # Drive-qualified relative path: not absolute() but escapes on Windows.
+        ("{p}", {"p": r"C:Windows\evil"}, "Absolute paths"),
+        # Adjacent placeholders can concatenate into ".." even when each value
+        # alone looks safe.
+        ("{a}{b}", {"a": ".", "b": "."}, "Path traversal"),
+    ],
+)
+def test_interpolate_output_file_rejects_unsafe_inputs(
+    template, malicious_inputs, expected_error
+):
+    """Untrusted inputs must not escape the output_file path via interpolation."""
+    task = Task(
+        description="d",
+        expected_output="e",
+        output_file=template,
+    )
+    with pytest.raises(ValueError, match=expected_error):
+        task.interpolate_inputs_and_add_conversation_history(inputs=malicious_inputs)
+
+
+def test_interpolate_output_file_allows_safe_inputs(tmp_path):
+    """Safe input values and developer-chosen absolute base paths still work."""
+    task = Task(
+        description="d",
+        expected_output="e",
+        output_file="reports/{name}.md",
+    )
+    task.interpolate_inputs_and_add_conversation_history(inputs={"name": "q3_summary"})
+    assert task.output_file == "reports/q3_summary.md"
+
+    abs_task = Task(
+        description="d",
+        expected_output="e",
+        output_file=str(tmp_path / "{topic}" / "out.md"),
+    )
+    abs_task.interpolate_inputs_and_add_conversation_history(inputs={"topic": "sales"})
+    assert abs_task.output_file == str(tmp_path / "sales" / "out.md")
+
+
 def test_interpolate_only():
     """Test the interpolate_only method for various scenarios including JSON structure preservation."""