Fix event listener to pass event parameter for enhanced error display

- Updated on_llm_call_failed to pass the event object to handle_llm_call_failed
- This enables the console formatter to display enhanced error details
- Completes the integration of enhanced error handling for custom endpoints

Co-Authored-By: Jo\u00E3o <joao@crewai.com>

docs/docs.json

@@ -9,12 +9,7 @@
   },
   "favicon": "/images/favicon.svg",
   "contextual": {
-    "options": [
-      "copy",
-      "view",
-      "chatgpt",
-      "claude"
-    ]
+    "options": ["copy", "view", "chatgpt", "claude"]
   },
   "navigation": {
     "languages": [
@@ -55,32 +50,22 @@
             "groups": [
               {
                 "group": "Get Started",
-                "pages": [
-                  "en/introduction",
-                  "en/installation",
-                  "en/quickstart"
-                ]
+                "pages": ["en/introduction", "en/installation", "en/quickstart"]
               },
               {
                 "group": "Guides",
                 "pages": [
                   {
                     "group": "Strategy",
-                    "pages": [
-                      "en/guides/concepts/evaluating-use-cases"
-                    ]
+                    "pages": ["en/guides/concepts/evaluating-use-cases"]
                   },
                   {
                     "group": "Agents",
-                    "pages": [
-                      "en/guides/agents/crafting-effective-agents"
-                    ]
+                    "pages": ["en/guides/agents/crafting-effective-agents"]
                   },
                   {
                     "group": "Crews",
-                    "pages": [
-                      "en/guides/crews/first-crew"
-                    ]
+                    "pages": ["en/guides/crews/first-crew"]
                   },
                   {
                     "group": "Flows",
@@ -94,7 +79,6 @@
                     "pages": [
                       "en/guides/advanced/customizing-prompts",
                       "en/guides/advanced/fingerprinting"
-
                     ]
                   }
                 ]
@@ -241,6 +225,7 @@
                   "en/observability/langtrace",
                   "en/observability/maxim",
                   "en/observability/mlflow",
+                  "en/observability/neatlogs",
                   "en/observability/openlit",
                   "en/observability/opik",
                   "en/observability/patronus-evaluation",
@@ -274,9 +259,7 @@
               },
               {
                 "group": "Telemetry",
-                "pages": [
-                  "en/telemetry"
-                ]
+                "pages": ["en/telemetry"]
               }
             ]
           },
@@ -285,9 +268,7 @@
             "groups": [
               {
                 "group": "Getting Started",
-                "pages": [
-                  "en/enterprise/introduction"
-                ]
+                "pages": ["en/enterprise/introduction"]
               },
               {
                 "group": "Features",
@@ -342,9 +323,7 @@
               },
               {
                 "group": "Resources",
-                "pages": [
-                  "en/enterprise/resources/frequently-asked-questions"
-                ]
+                "pages": ["en/enterprise/resources/frequently-asked-questions"]
               }
             ]
           },
@@ -353,9 +332,7 @@
             "groups": [
               {
                 "group": "Getting Started",
-                "pages": [
-                  "en/api-reference/introduction"
-                ]
+                "pages": ["en/api-reference/introduction"]
               },
               {
                 "group": "Endpoints",
@@ -365,16 +342,13 @@
           },
           {
             "tab": "Examples",
-                        "groups": [
+            "groups": [
               {
                 "group": "Examples",
-                "pages": [
-                  "en/examples/example"
-                ]
+                "pages": ["en/examples/example"]
               }
             ]
           }
-
         ]
       },
       {
@@ -425,21 +399,15 @@
                 "pages": [
                   {
                     "group": "Estratégia",
-                    "pages": [
-                      "pt-BR/guides/concepts/evaluating-use-cases"
-                    ]
+                    "pages": ["pt-BR/guides/concepts/evaluating-use-cases"]
                   },
                   {
                     "group": "Agentes",
-                    "pages": [
-                      "pt-BR/guides/agents/crafting-effective-agents"
-                    ]
+                    "pages": ["pt-BR/guides/agents/crafting-effective-agents"]
                   },
                   {
                     "group": "Crews",
-                    "pages": [
-                      "pt-BR/guides/crews/first-crew"
-                    ]
+                    "pages": ["pt-BR/guides/crews/first-crew"]
                   },
                   {
                     "group": "Flows",
@@ -632,9 +600,7 @@
               },
               {
                 "group": "Telemetria",
-                "pages": [
-                  "pt-BR/telemetry"
-                ]
+                "pages": ["pt-BR/telemetry"]
               }
             ]
           },
@@ -643,9 +609,7 @@
             "groups": [
               {
                 "group": "Começando",
-                "pages": [
-                  "pt-BR/enterprise/introduction"
-                ]
+                "pages": ["pt-BR/enterprise/introduction"]
               },
               {
                 "group": "Funcionalidades",
@@ -710,9 +674,7 @@
             "groups": [
               {
                 "group": "Começando",
-                "pages": [
-                  "pt-BR/api-reference/introduction"
-                ]
+                "pages": ["pt-BR/api-reference/introduction"]
               },
               {
                 "group": "Endpoints",
@@ -722,16 +684,13 @@
           },
           {
             "tab": "Exemplos",
-                        "groups": [
+            "groups": [
               {
                 "group": "Exemplos",
-                "pages": [
-                  "pt-BR/examples/example"
-                ]
+                "pages": ["pt-BR/examples/example"]
               }
             ]
           }
-
         ]
       }
     ]

docs/en/observability/neatlogs.mdx

@@ -0,0 +1,140 @@
+---
+title: Neatlogs Integration
+description: Understand, debug, and share your CrewAI agent runs
+icon: magnifying-glass-chart
+---
+
+# Introduction
+
+Neatlogs helps you **see what your agent did**, **why**, and **share it**.
+
+It captures every step: thoughts, tool calls, responses, evaluations. No raw logs. Just clear, structured traces. Great for debugging and collaboration.
+
+---
+
+## Why use Neatlogs?
+
+CrewAI agents use multiple tools and reasoning steps. When something goes wrong, you need context — not just errors.
+
+Neatlogs lets you:
+
+- Follow the full decision path
+- Add feedback directly on steps
+- Chat with the trace using AI assistant
+- Share runs publicly for feedback
+- Turn insights into tasks
+
+All in one place.
+
+Manage your traces effortlessly
+
+![Traces](/images/neatlogs-1.png)
+![Trace Response](/images/neatlogs-2.png)
+
+The best UX to view a CrewAI trace. Post comments anywhere you want. Use AI to debug.
+
+![Trace Details](/images/neatlogs-3.png)
+![Ai Chat Bot With A Trace](/images/neatlogs-4.png)
+![Comments Drawer](/images/neatlogs-5.png)
+
+---
+
+## Core Features
+
+- **Trace Viewer**: Track thoughts, tools, and decisions in sequence
+- **Inline Comments**: Tag teammates on any trace step
+- **Feedback & Evaluation**: Mark outputs as correct or incorrect
+- **Error Highlighting**: Automatic flagging of API/tool failures
+- **Task Conversion**: Convert comments into assigned tasks
+- **Ask the Trace (AI)**: Chat with your trace using Neatlogs AI bot
+- **Public Sharing**: Publish trace links to your community
+
+---
+
+## Quick Setup with CrewAI
+
+<Steps>
+  <Step title="Sign Up & Get API Key">
+    Visit [neatlogs.com](https://neatlogs.com/?utm_source=crewAI-docs), create a project, copy the API key.
+  </Step>
+  <Step title="Install SDK">
+    ```bash
+    pip install neatlogs
+    ```
+    (Latest version 0.8.0, Python 3.8+; MIT license) :contentReference[oaicite:1]{index=1}
+  </Step>
+  <Step title="Initialize Neatlogs">
+    Before starting Crew agents, add:
+
+    ```python
+    import neatlogs
+    neatlogs.init("YOUR_PROJECT_API_KEY")
+    ```
+
+    Agents run as usual. Neatlogs captures everything automatically.
+
+  </Step>
+</Steps>
+
+---
+
+## Under the Hood
+
+According to GitHub, Neatlogs:
+
+- Captures thoughts, tool calls, responses, errors, and token stats :contentReference[oaicite:2]{index=2}
+- Supports AI-powered task generation and robust evaluation workflows :contentReference[oaicite:3]{index=3}
+
+All with just two lines of code.
+
+---
+
+## Watch It Work
+
+### 🔍 Full Demo (4 min)
+
+<iframe
+  width="100%"
+  height="315"
+  src="https://www.youtube.com/embed/8KDme9T2I7Q?si=b8oHteaBwFNs_Duk"
+  title="YouTube video player"
+  frameBorder="0"
+  allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture"
+  allowFullScreen
+></iframe>
+
+### ⚙️ CrewAI Integration (30 s)
+
+<iframe
+  className="w-full aspect-video rounded-xl"
+  src="https://www.loom.com/embed/9c78b552af43452bb3e4783cb8d91230?sid=e9d7d370-a91a-49b0-809e-2f375d9e801d"
+  title="Loom video player"
+  frameBorder="0"
+  allowFullScreen
+></iframe>
+
+---
+
+## Links & Support
+
+- 📘 [Neatlogs Docs](https://docs.neatlogs.com/)
+- 🔐 [Dashboard & API Key](https://app.neatlogs.com/)
+- 🐦 [Follow on Twitter](https://twitter.com/neatlogs)
+- 📧 Contact: hello@neatlogs.com
+- 🛠 [GitHub SDK](https://github.com/NeatLogs/neatlogs) :contentReference[oaicite:4]{index=4}
+
+---
+
+## TL;DR
+
+With just:
+
+```bash
+pip install neatlogs
+
+import neatlogs
+neatlogs.init("YOUR_API_KEY")
+
+You can now capture, understand, share, and act on your CrewAI agent runs in seconds.
+No setup overhead. Full trace transparency. Full team collaboration.
+```

reproduce_issue_3165.py

@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+"""
+Reproduction script for issue #3165: LLM Failed with Custom OpenAI-Compatible Endpoint
+
+This script reproduces the bug where CrewAI shows generic "LLM Failed" errors
+instead of propagating specific error details from custom endpoints.
+"""
+
+import sys
+from crewai import Agent, Task, Crew
+from crewai.llm import LLM
+
+def test_custom_endpoint_error_handling():
+    """Test error handling with a custom OpenAI-compatible endpoint."""
+    
+    print("Testing custom endpoint error handling...")
+    
+    custom_llm = LLM(
+        model="gpt-3.5-turbo",
+        base_url="https://non-existent-endpoint.example.com/v1",
+        api_key="fake-api-key-for-testing"
+    )
+    
+    agent = Agent(
+        role="Test Agent",
+        goal="Test custom endpoint error handling",
+        backstory="A test agent for reproducing issue #3165",
+        llm=custom_llm,
+        verbose=True
+    )
+    
+    task = Task(
+        description="Say hello world",
+        expected_output="A simple greeting",
+        agent=agent
+    )
+    
+    crew = Crew(
+        agents=[agent],
+        tasks=[task],
+        verbose=True
+    )
+    
+    try:
+        print("\nAttempting to run crew with custom endpoint...")
+        result = crew.kickoff()
+        print(f"Unexpected success: {result}")
+    except Exception as e:
+        print(f"\nCaught exception: {type(e).__name__}")
+        print(f"Exception message: {str(e)}")
+        
+        if "LLM Failed" in str(e) and "connection" not in str(e).lower():
+            print("\n❌ BUG CONFIRMED: Generic 'LLM Failed' error without specific details")
+            print("Expected: Specific connection/authentication error details")
+            return False
+        else:
+            print("\n✅ Good: Specific error details preserved")
+            return True
+
+def test_direct_llm_call():
+    """Test direct LLM call with custom endpoint."""
+    
+    print("\n" + "="*60)
+    print("Testing direct LLM call with custom endpoint...")
+    
+    custom_llm = LLM(
+        model="gpt-3.5-turbo", 
+        base_url="https://non-existent-endpoint.example.com/v1",
+        api_key="fake-api-key-for-testing"
+    )
+    
+    try:
+        print("Attempting direct LLM call...")
+        response = custom_llm.call("Hello world")
+        print(f"Unexpected success: {response}")
+    except Exception as e:
+        print(f"\nCaught exception: {type(e).__name__}")
+        print(f"Exception message: {str(e)}")
+        
+        error_msg = str(e).lower()
+        if any(keyword in error_msg for keyword in ["connection", "resolve", "network", "timeout", "unreachable"]):
+            print("\n✅ Good: Specific connection error details preserved")
+            return True
+        else:
+            print("\n❌ BUG CONFIRMED: Generic error without connection details")
+            print("Expected: Specific connection error details")
+            return False
+
+if __name__ == "__main__":
+    print("Reproducing issue #3165: LLM Failed with Custom OpenAI-Compatible Endpoint")
+    print("="*80)
+    
+    crew_test_passed = test_custom_endpoint_error_handling()
+    direct_test_passed = test_direct_llm_call()
+    
+    print("\n" + "="*80)
+    print("SUMMARY:")
+    print(f"Crew-level test: {'PASSED' if crew_test_passed else 'FAILED (bug confirmed)'}")
+    print(f"Direct LLM test: {'PASSED' if direct_test_passed else 'FAILED (bug confirmed)'}")
+    
+    if not crew_test_passed or not direct_test_passed:
+        print("\n❌ Issue #3165 reproduced successfully")
+        print("CrewAI is showing generic errors instead of specific endpoint error details")
+        sys.exit(1)
+    else:
+        print("\n✅ Issue #3165 appears to be fixed")
+        sys.exit(0)

src/crewai/llm.py

@@ -984,10 +984,27 @@ class LLM(BaseLLM):
                 # whether to summarize the content or abort based on the respect_context_window flag
                 raise
             except Exception as e:
+                error_info = {
+                    "error_type": type(e).__name__,
+                    "original_error": str(e),
+                    "endpoint_info": {
+                        "base_url": self.base_url,
+                        "model": self.model,
+                        "api_base": self.api_base,
+                    } if self.base_url or self.api_base else None
+                }
+                
                 assert hasattr(crewai_event_bus, "emit")
                 crewai_event_bus.emit(
                     self,
-                    event=LLMCallFailedEvent(error=str(e), from_task=from_task, from_agent=from_agent),
+                    event=LLMCallFailedEvent(
+                        error=str(e),
+                        error_type=error_info["error_type"],
+                        original_error=error_info["original_error"],
+                        endpoint_info=error_info["endpoint_info"],
+                        from_task=from_task, 
+                        from_agent=from_agent
+                    ),
                 )
                 logging.error(f"LiteLLM call failed: {str(e)}")
                 raise

src/crewai/utilities/events/crewai_event_bus.py

@@ -1,6 +1,6 @@
 import threading
 from contextlib import contextmanager
-from typing import Any, Callable, Type, TypeVar, cast
+from typing import Any, Callable, Dict, List, Type, TypeVar, cast
 
 from blinker import Signal
 
@@ -14,13 +14,10 @@ class CrewAIEventsBus:
     """
     A singleton event bus that uses blinker signals for event handling.
     Allows both internal (Flow/Crew) and external event handling.
-    Handlers are global by default for cross-thread communication,
-    with optional thread-local isolation for testing scenarios.
     """
 
     _instance = None
     _lock = threading.Lock()
-    _thread_local: threading.local = threading.local()
 
     def __new__(cls):
         if cls._instance is None:
@@ -33,46 +30,7 @@ class CrewAIEventsBus:
     def _initialize(self) -> None:
         """Initialize the event bus internal state"""
         self._signal = Signal("crewai_event_bus")
-        self._global_handlers: dict[type[BaseEvent], list[Callable]] = {}
-
-    @property
-    def _handlers(self) -> dict[type[BaseEvent], list[Callable]]:
-        if not hasattr(CrewAIEventsBus._thread_local, "handlers"):
-            CrewAIEventsBus._thread_local.handlers = {}
-        return CrewAIEventsBus._thread_local.handlers
-
-    @_handlers.setter
-    def _handlers(self, value: dict[type[BaseEvent], list[Callable]]) -> None:
-        if not hasattr(CrewAIEventsBus._thread_local, "handlers"):
-            CrewAIEventsBus._thread_local.handlers = {}
-        CrewAIEventsBus._thread_local.handlers = value
-
-    def _add_handler_with_deduplication(
-        self, handlers_dict: dict, event_type: Type[BaseEvent], handler: Callable
-    ) -> bool:
-        """
-        Add a handler to the specified handlers dictionary with deduplication.
-
-        Args:
-            handlers_dict: The dictionary to add the handler to
-            event_type: The event type
-            handler: The handler function to add
-
-        Returns:
-            bool: True if handler was added, False if it was already present
-        """
-        if event_type not in handlers_dict:
-            handlers_dict[event_type] = []
-
-        # Check if handler is already registered
-        for existing_handler in handlers_dict[event_type]:
-            if existing_handler is handler:
-                # Handler already exists, don't add duplicate
-                return False
-
-        # Add the handler
-        handlers_dict[event_type].append(handler)
-        return True
+        self._handlers: Dict[Type[BaseEvent], List[Callable]] = {}
 
     def on(
         self, event_type: Type[EventT]
@@ -80,13 +38,6 @@ class CrewAIEventsBus:
         """
         Decorator to register an event handler for a specific event type.
 
-        Handlers registered with this decorator are global by default,
-        allowing cross-thread event communication. Use scoped_handlers()
-        for thread-local isolation in testing scenarios.
-
-        Duplicate handlers are automatically prevented - the same handler
-        function will only be registered once per event type.
-
         Usage:
             @crewai_event_bus.on(AgentExecutionCompletedEvent)
             def on_agent_execution_completed(
@@ -99,38 +50,23 @@ class CrewAIEventsBus:
         def decorator(
             handler: Callable[[Any, EventT], None],
         ) -> Callable[[Any, EventT], None]:
-            was_added = self._add_handler_with_deduplication(
-                self._global_handlers, event_type, handler
+            if event_type not in self._handlers:
+                self._handlers[event_type] = []
+            self._handlers[event_type].append(
+                cast(Callable[[Any, EventT], None], handler)
             )
-            if not was_added:
-                # Log that duplicate was prevented (optional)
-                print(
-                    f"[EventBus Info] Handler '{handler.__name__}' already registered for {event_type.__name__}"
-                )
             return handler
 
         return decorator
 
     def emit(self, source: Any, event: BaseEvent) -> None:
         """
-        Emit an event to all registered handlers (both global and thread-local)
+        Emit an event to all registered handlers
 
         Args:
             source: The object emitting the event
             event: The event instance to emit
         """
-        # Call global handlers (default behavior, cross-thread)
-        for event_type, handlers in self._global_handlers.items():
-            if isinstance(event, event_type):
-                for handler in handlers:
-                    try:
-                        handler(source, event)
-                    except Exception as e:
-                        print(
-                            f"[EventBus Error] Global handler '{handler.__name__}' failed for event '{event_type.__name__}': {e}"
-                        )
-
-        # Call thread-local handlers (for testing isolation)
         for event_type, handlers in self._handlers.items():
             if isinstance(event, event_type):
                 for handler in handlers:
@@ -138,76 +74,32 @@ class CrewAIEventsBus:
                         handler(source, event)
                     except Exception as e:
                         print(
-                            f"[EventBus Error] Thread-local handler '{handler.__name__}' failed for event '{event_type.__name__}': {e}"
+                            f"[EventBus Error] Handler '{handler.__name__}' failed for event '{event_type.__name__}': {e}"
                         )
 
-        # Send to blinker signal (existing mechanism)
         self._signal.send(source, event=event)
 
     def register_handler(
-        self, event_type: Type[BaseEvent], handler: Callable[[Any, BaseEvent], None]
-    ) -> bool:
-        """
-        Register an event handler for a specific event type (global)
-
-        Args:
-            event_type: The event type to handle
-            handler: The handler function to register
-
-        Returns:
-            bool: True if handler was added, False if it was already present
-        """
-        return self._add_handler_with_deduplication(
-            self._global_handlers, event_type, handler
+        self, event_type: Type[EventTypes], handler: Callable[[Any, EventTypes], None]
+    ) -> None:
+        """Register an event handler for a specific event type"""
+        if event_type not in self._handlers:
+            self._handlers[event_type] = []
+        self._handlers[event_type].append(
+            cast(Callable[[Any, EventTypes], None], handler)
         )
 
-    def unregister_handler(
-        self, event_type: Type[BaseEvent], handler: Callable[[Any, BaseEvent], None]
-    ) -> bool:
-        """
-        Unregister an event handler for a specific event type (global)
-
-        Args:
-            event_type: The event type
-            handler: The handler function to unregister
-
-        Returns:
-            bool: True if handler was removed, False if it wasn't found
-        """
-        if event_type in self._global_handlers:
-            try:
-                self._global_handlers[event_type].remove(handler)
-                return True
-            except ValueError:
-                return False
-        return False
-
-    def get_handler_count(self, event_type: Type[BaseEvent]) -> int:
-        """
-        Get the number of handlers registered for a specific event type
-
-        Args:
-            event_type: The event type to check
-
-        Returns:
-            int: Number of handlers registered for this event type
-        """
-        return len(self._global_handlers.get(event_type, []))
-
     @contextmanager
     def scoped_handlers(self):
         """
-        Context manager for temporary thread-local event handling scope.
-        Useful for testing or temporary event handling with thread isolation.
-
-        This creates thread-local handlers that are isolated from global handlers,
-        making it useful for testing scenarios where you want to avoid interference.
+        Context manager for temporary event handling scope.
+        Useful for testing or temporary event handling.
 
         Usage:
             with crewai_event_bus.scoped_handlers():
                 @crewai_event_bus.on(CrewKickoffStarted)
                 def temp_handler(source, event):
-                    print("Temporary thread-local handler")
+                    print("Temporary handler")
                 # Do stuff...
             # Handlers are cleared after the context
         """
@@ -218,25 +110,6 @@ class CrewAIEventsBus:
         finally:
             self._handlers = previous_handlers
 
-    @contextmanager
-    def scoped_global_handlers(self):
-        """
-        Context manager for temporary global event handling scope.
-        Useful for testing or temporary global event handling.
-
-        Usage:
-            with crewai_event_bus.scoped_global_handlers():
-                crewai_event_bus.register_handler(CrewKickoffStarted, temp_handler)
-                # Do stuff...
-            # Global handlers are cleared after the context
-        """
-        previous_global_handlers = self._global_handlers.copy()
-        self._global_handlers.clear()
-        try:
-            yield
-        finally:
-            self._global_handlers = previous_global_handlers
-
 
 # Global instance
 crewai_event_bus = CrewAIEventsBus()

src/crewai/utilities/events/event_listener.py

@@ -361,6 +361,7 @@ class EventListener(BaseEventListener):
                 self.formatter.current_tool_branch,
                 event.error,
                 self.formatter.current_crew_tree,
+                event,
             )
 
         @crewai_event_bus.on(LLMStreamChunkEvent)

src/crewai/utilities/events/llm_events.py

@@ -67,6 +67,9 @@ class LLMCallFailedEvent(LLMEventBase):
 
     error: str
     type: str = "llm_call_failed"
+    error_type: Optional[str] = None
+    original_error: Optional[str] = None
+    endpoint_info: Optional[Dict[str, Any]] = None
 
 
 class FunctionCall(BaseModel):

src/crewai/utilities/events/utils/console_formatter.py

@@ -721,7 +721,7 @@ class ConsoleFormatter:
                 self.print()
 
     def handle_llm_call_failed(
-        self, tool_branch: Optional[Tree], error: str, crew_tree: Optional[Tree]
+        self, tool_branch: Optional[Tree], error: str, crew_tree: Optional[Tree], event: Optional[Any] = None
     ) -> None:
         """Handle LLM call failed event."""
         if not self.verbose:
@@ -764,9 +764,19 @@ class ConsoleFormatter:
                 self.print(tree_to_use)
                 self.print()
 
-        # Show error panel
+        # Show detailed error panel
         error_content = Text()
         error_content.append("❌ LLM Call Failed\n", style="red bold")
+        
+        if event and hasattr(event, 'error_type') and event.error_type:
+            error_content.append(f"Error Type: {event.error_type}\n", style="yellow")
+        
+        if event and hasattr(event, 'endpoint_info') and event.endpoint_info:
+            endpoint = event.endpoint_info.get('base_url') or event.endpoint_info.get('api_base')
+            if endpoint:
+                error_content.append(f"Endpoint: {endpoint}\n", style="cyan")
+                error_content.append(f"Model: {event.endpoint_info.get('model', 'unknown')}\n", style="cyan")
+        
         error_content.append("Error: ", style="white")
         error_content.append(str(error), style="red")
 

tests/test_custom_endpoint_error_handling.py

@@ -0,0 +1,185 @@
+"""
+Tests for custom endpoint error handling (issue #3165).
+
+These tests verify that CrewAI properly propagates specific error details
+from custom OpenAI-compatible endpoints instead of showing generic "LLM Failed" errors.
+"""
+
+import pytest
+from unittest.mock import patch, MagicMock
+from crewai.llm import LLM
+from crewai.utilities.events.llm_events import LLMCallFailedEvent
+from crewai.utilities.events.utils.console_formatter import ConsoleFormatter
+import requests
+
+
+class TestCustomEndpointErrorHandling:
+    """Test error handling for custom OpenAI-compatible endpoints."""
+
+    def test_connection_error_preserves_details(self):
+        """Test that connection errors preserve specific error details."""
+        custom_llm = LLM(
+            model="gpt-3.5-turbo",
+            base_url="https://non-existent-endpoint.example.com/v1",
+            api_key="fake-api-key"
+        )
+        
+        with patch('litellm.completion') as mock_completion:
+            mock_completion.side_effect = requests.exceptions.ConnectionError(
+                "Failed to establish a new connection: [Errno -2] Name or service not known"
+            )
+            
+            with pytest.raises(requests.exceptions.ConnectionError) as exc_info:
+                custom_llm.call("Hello world")
+            
+            assert "Name or service not known" in str(exc_info.value)
+
+    def test_authentication_error_preserves_details(self):
+        """Test that authentication errors preserve specific error details."""
+        custom_llm = LLM(
+            model="gpt-3.5-turbo",
+            base_url="https://api.openai.com/v1",
+            api_key="invalid-api-key"
+        )
+        
+        with patch('litellm.completion') as mock_completion:
+            mock_completion.side_effect = Exception(
+                "AuthenticationError: Incorrect API key provided"
+            )
+            
+            with pytest.raises(Exception) as exc_info:
+                custom_llm.call("Hello world")
+            
+            assert "AuthenticationError" in str(exc_info.value)
+            assert "Incorrect API key" in str(exc_info.value)
+
+    def test_llm_call_failed_event_enhanced_fields(self):
+        """Test that LLMCallFailedEvent includes enhanced error information."""
+        custom_llm = LLM(
+            model="gpt-3.5-turbo",
+            base_url="https://custom-endpoint.example.com/v1",
+            api_key="test-key"
+        )
+        
+        captured_events = []
+        
+        def capture_event(sender, event):
+            captured_events.append(event)
+        
+        with patch('crewai.utilities.events.crewai_event_bus.crewai_event_bus.emit', side_effect=capture_event):
+            with patch('litellm.completion') as mock_completion:
+                mock_completion.side_effect = requests.exceptions.ConnectionError(
+                    "Connection failed"
+                )
+                
+                with pytest.raises(requests.exceptions.ConnectionError):
+                    custom_llm.call("Hello world")
+        
+        assert len(captured_events) == 2  # Started and Failed events
+        failed_event = captured_events[1]
+        assert isinstance(failed_event, LLMCallFailedEvent)
+        assert failed_event.error_type == "ConnectionError"
+        assert failed_event.original_error == "Connection failed"
+        assert failed_event.endpoint_info is not None
+        assert failed_event.endpoint_info["base_url"] == "https://custom-endpoint.example.com/v1"
+        assert failed_event.endpoint_info["model"] == "gpt-3.5-turbo"
+
+    def test_console_formatter_displays_enhanced_error_info(self):
+        """Test that console formatter displays enhanced error information."""
+        formatter = ConsoleFormatter(verbose=True)
+        
+        mock_event = MagicMock()
+        mock_event.error_type = "ConnectionError"
+        mock_event.endpoint_info = {
+            "base_url": "https://custom-endpoint.example.com/v1",
+            "model": "gpt-3.5-turbo"
+        }
+        
+        captured_output = []
+        
+        def mock_print_panel(content, title, style):
+            captured_output.append(str(content))
+        
+        formatter.print_panel = mock_print_panel
+        
+        formatter.handle_llm_call_failed(
+            tool_branch=None,
+            error="Connection failed",
+            crew_tree=None,
+            event=mock_event
+        )
+        
+        output = captured_output[0]
+        assert "Error Type: ConnectionError" in output
+        assert "Endpoint: https://custom-endpoint.example.com/v1" in output
+        assert "Model: gpt-3.5-turbo" in output
+        assert "Connection failed" in output
+
+    def test_backward_compatibility_without_enhanced_fields(self):
+        """Test that console formatter works without enhanced fields for backward compatibility."""
+        formatter = ConsoleFormatter(verbose=True)
+        
+        captured_output = []
+        
+        def mock_print_panel(content, title, style):
+            captured_output.append(str(content))
+        
+        formatter.print_panel = mock_print_panel
+        
+        formatter.handle_llm_call_failed(
+            tool_branch=None,
+            error="Generic error message",
+            crew_tree=None,
+            event=None
+        )
+        
+        output = captured_output[0]
+        assert "❌ LLM Call Failed" in output
+        assert "Generic error message" in output
+        assert "Error Type:" not in output
+        assert "Endpoint:" not in output
+
+    def test_streaming_response_error_handling(self):
+        """Test that streaming responses also preserve error details."""
+        custom_llm = LLM(
+            model="gpt-3.5-turbo",
+            base_url="https://custom-endpoint.example.com/v1",
+            api_key="test-key",
+            stream=True
+        )
+        
+        with patch('litellm.completion') as mock_completion:
+            mock_completion.side_effect = requests.exceptions.ConnectionError(
+                "Streaming connection failed"
+            )
+            
+            with pytest.raises(Exception) as exc_info:
+                custom_llm.call("Hello world")
+            
+            assert "Streaming connection failed" in str(exc_info.value)
+
+    def test_non_custom_endpoint_error_handling(self):
+        """Test that standard OpenAI endpoint errors are handled normally."""
+        standard_llm = LLM(
+            model="gpt-3.5-turbo",
+            api_key="test-key"
+        )
+        
+        captured_events = []
+        
+        def capture_event(sender, event):
+            captured_events.append(event)
+        
+        with patch('crewai.utilities.events.crewai_event_bus.crewai_event_bus.emit', side_effect=capture_event):
+            with patch('litellm.completion') as mock_completion:
+                mock_completion.side_effect = Exception("Standard API error")
+                
+                with pytest.raises(Exception):
+                    standard_llm.call("Hello world")
+        
+        assert len(captured_events) == 2  # Started and Failed events
+        failed_event = captured_events[1]
+        assert isinstance(failed_event, LLMCallFailedEvent)
+        assert failed_event.error_type == "Exception"
+        assert failed_event.original_error == "Standard API error"
+        assert failed_event.endpoint_info is None  # No custom endpoint info

tests/utilities/events/test_crewai_event_bus.py

@@ -1,31 +1,13 @@
-import threading
-from typing import Any, Callable, cast
 from unittest.mock import Mock
 
-import pytest
-
 from crewai.utilities.events.base_events import BaseEvent
 from crewai.utilities.events.crewai_event_bus import crewai_event_bus
 
 
-@pytest.fixture(autouse=True)
-def scoped_event_handlers():
-    with crewai_event_bus.scoped_handlers():
-        yield
-
-
 class TestEvent(BaseEvent):
     pass
 
 
-class AnotherThreadTestEvent(BaseEvent):
-    pass
-
-
-class CrossThreadTestEvent(BaseEvent):
-    pass
-
-
 def test_specific_event_handler():
     mock_handler = Mock()
 
@@ -62,444 +44,4 @@ def test_event_bus_error_handling(capfd):
 
     out, err = capfd.readouterr()
     assert "Simulated handler failure" in out
-    assert "Global handler 'broken_handler' failed" in out
-
-
-def test_singleton_pattern_across_threads():
-    instances = []
-
-    def get_instance():
-        instances.append(crewai_event_bus)
-
-    threads = []
-    for _ in range(10):
-        thread = threading.Thread(target=get_instance)
-        threads.append(thread)
-        thread.start()
-
-    for thread in threads:
-        thread.join()
-    assert len(instances) == 10
-    for instance in instances:
-        assert instance is crewai_event_bus
-        assert instance is instances[0]
-
-
-def test_default_handlers_are_global():
-    """Test that handlers registered with @crewai_event_bus.on() are global by default."""
-    received_events = []
-    mock_handler = Mock()
-
-    @crewai_event_bus.on(CrossThreadTestEvent)
-    def global_handler(source, event):
-        received_events.append((source, event))
-        mock_handler(source, event)
-
-    def thread_worker(thread_id):
-        # Emit event from a different thread
-        event = CrossThreadTestEvent(type=f"cross_thread_event_{thread_id}")
-        crewai_event_bus.emit(f"thread_source_{thread_id}", event)
-
-    # Start multiple threads that emit events
-    threads = []
-    for i in range(3):
-        thread = threading.Thread(target=thread_worker, args=(i,))
-        threads.append(thread)
-        thread.start()
-
-    for thread in threads:
-        thread.join()
-
-    # Verify that the global handler received all events from different threads
-    assert len(received_events) == 3
-    assert mock_handler.call_count == 3
-
-    # Check that events from different threads were received
-    for i in range(3):
-        source, event = received_events[i]
-        assert source == f"thread_source_{i}"
-        assert event.type == f"cross_thread_event_{i}"
-
-
-def test_scoped_handlers_thread_isolation():
-    """Test that scoped_handlers() provides thread-local isolation for testing."""
-    global_events = []
-    scoped_events = []
-
-    # Register a global handler
-    @crewai_event_bus.on(CrossThreadTestEvent)
-    def global_handler(source, event):
-        global_events.append((source, event))
-
-    # Emit an event - should be received by global handler
-    event1 = CrossThreadTestEvent(type="event_1")
-    crewai_event_bus.emit("source_1", event1)
-    assert len(global_events) == 1
-
-    # Use scoped handlers for testing isolation
-    with crewai_event_bus.scoped_handlers():
-        # Register a handler in the scoped context (thread-local)
-        @crewai_event_bus.on(CrossThreadTestEvent)
-        def scoped_handler(source, event):
-            scoped_events.append((source, event))
-
-        # Emit event - should be received by scoped handler only
-        event2 = CrossThreadTestEvent(type="event_2")
-        crewai_event_bus.emit("source_2", event2)
-
-    # After scope, emit another event - should be received by global handler only
-    event3 = CrossThreadTestEvent(type="event_3")
-    crewai_event_bus.emit("source_3", event3)
-
-    # Verify events
-    assert len(global_events) == 2  # event_1 and event_3
-    assert len(scoped_events) == 1  # only event_2
-    assert global_events[0] == ("source_1", event1)
-    assert scoped_events[0] == ("source_2", event2)
-    assert global_events[1] == ("source_3", event3)
-
-
-def test_scoped_handlers_thread_safety():
-    """Test that scoped handlers work correctly across multiple threads."""
-    thread_results = {}
-
-    def thread_worker(thread_id):
-        with crewai_event_bus.scoped_handlers():
-            mock_handler = Mock()
-
-            @crewai_event_bus.on(AnotherThreadTestEvent)
-            def scoped_handler(source, event):
-                mock_handler(f"scoped_thread_{thread_id}", event)
-
-            scoped_event = AnotherThreadTestEvent(type=f"scoped_event_{thread_id}")
-            crewai_event_bus.emit(f"scoped_source_{thread_id}", scoped_event)
-
-            thread_results[thread_id] = {
-                "mock_handler": mock_handler,
-                "scoped_event": scoped_event,
-            }
-
-        # After scope, emit event - should not be received by scoped handler
-        post_scoped_event = AnotherThreadTestEvent(type=f"post_scoped_{thread_id}")
-        crewai_event_bus.emit(f"post_source_{thread_id}", post_scoped_event)
-
-    threads = []
-    for i in range(5):
-        thread = threading.Thread(target=thread_worker, args=(i,))
-        threads.append(thread)
-        thread.start()
-
-    for thread in threads:
-        thread.join()
-
-    for thread_id, result in thread_results.items():
-        result["mock_handler"].assert_called_once_with(
-            f"scoped_thread_{thread_id}", result["scoped_event"]
-        )
-
-
-def test_register_handler_method():
-    """Test the register_handler method works with global handlers."""
-    received_events = []
-
-    def handler(source, event):
-        received_events.append((source, event))
-
-    # Register handler using the method
-    crewai_event_bus.register_handler(CrossThreadTestEvent, handler)
-
-    # Emit event from different thread
-    def thread_worker():
-        event = CrossThreadTestEvent(type="test_event")
-        crewai_event_bus.emit("thread_source", event)
-
-    thread = threading.Thread(target=thread_worker)
-    thread.start()
-    thread.join()
-
-    # Verify handler received the event
-    assert len(received_events) == 1
-    assert received_events[0] == (
-        "thread_source",
-        CrossThreadTestEvent(type="test_event"),
-    )
-
-
-def test_scoped_global_handlers():
-    """Test the scoped_global_handlers context manager."""
-    global_events = []
-
-    def global_handler(source, event):
-        global_events.append((source, event))
-
-    # Register a global handler
-    crewai_event_bus.register_handler(CrossThreadTestEvent, global_handler)
-
-    # Emit an event - should be received
-    event1 = CrossThreadTestEvent(type="event_1")
-    crewai_event_bus.emit("source_1", event1)
-    assert len(global_events) == 1
-
-    # Use scoped global handlers
-    with crewai_event_bus.scoped_global_handlers():
-        # Register a different handler in scope
-        def scoped_handler(source, event):
-            global_events.append(("scoped", source, event))
-
-        crewai_event_bus.register_handler(CrossThreadTestEvent, scoped_handler)
-
-        # Emit event - should be received by scoped handler
-        event2 = CrossThreadTestEvent(type="event_2")
-        crewai_event_bus.emit("source_2", event2)
-
-    # After scope, original handler should be restored
-    event3 = CrossThreadTestEvent(type="event_3")
-    crewai_event_bus.emit("source_3", event3)
-
-    # Verify events
-    assert len(global_events) == 3
-    assert global_events[0] == ("source_1", event1)
-    assert global_events[1] == ("scoped", "source_2", event2)
-    assert global_events[2] == ("source_3", event3)
-
-
-def test_handler_duplication_scenarios():
-    """Test various scenarios where handler duplication can occur."""
-    call_counts = []
-
-    def handler(source, event):
-        call_counts.append(1)
-
-    # Scenario 1: Register the same handler multiple times
-    crewai_event_bus.register_handler(TestEvent, handler)
-    crewai_event_bus.register_handler(TestEvent, handler)  # Duplicate registration
-
-    # Scenario 2: Use decorator multiple times on the same function
-    @crewai_event_bus.on(TestEvent)
-    def decorated_handler1(source, event):
-        call_counts.append(1)
-
-    @crewai_event_bus.on(TestEvent)
-    def decorated_handler2(source, event):  # Same function name, different instance
-        call_counts.append(1)
-
-    # Emit an event
-    event = TestEvent(type="test_event")
-    crewai_event_bus.emit("source", event)
-
-    # Currently, all handlers are called (including duplicates)
-    # This shows the current behavior - handlers can be duplicated
-    assert len(call_counts) >= 4  # At least 4 calls (2 direct + 2 decorated)
-
-
-def test_module_reload_duplication():
-    """Test duplication that could occur from module reloading."""
-    call_counts = []
-
-    def create_handler():
-        def handler(source, event):
-            call_counts.append(1)
-
-        return handler
-
-    # Simulate module reload scenario
-    handler1 = create_handler()
-    handler2 = create_handler()  # Same function, different instance
-
-    crewai_event_bus.register_handler(TestEvent, handler1)
-    crewai_event_bus.register_handler(TestEvent, handler2)
-
-    event = TestEvent(type="test_event")
-    crewai_event_bus.emit("source", event)
-
-    # Both handlers are called (duplication)
-    assert len(call_counts) == 2
-
-
-def test_listener_class_duplication():
-    """Test duplication from multiple listener class instances."""
-    call_counts = []
-
-    class TestListener:
-        def __init__(self):
-            @crewai_event_bus.on(TestEvent)
-            def handler(source, event):
-                call_counts.append(1)
-
-    # Create multiple instances (simulating multiple imports)
-    listener1 = TestListener()
-    listener2 = TestListener()
-
-    event = TestEvent(type="test_event")
-    crewai_event_bus.emit("source", event)
-
-    # Both instances register handlers (duplication)
-    assert len(call_counts) == 2
-
-
-def test_handler_deduplication():
-    """Test that duplicate handlers are automatically prevented."""
-    call_counts = []
-
-    def handler(source, event):
-        call_counts.append(1)
-
-    # Register the same handler multiple times
-    result1 = crewai_event_bus.register_handler(TestEvent, handler)
-    result2 = crewai_event_bus.register_handler(
-        TestEvent, handler
-    )  # Duplicate registration
-
-    # First registration should succeed, second should fail
-    assert result1 is True
-    assert result2 is False
-
-    # Emit an event
-    event = TestEvent(type="test_event")
-    crewai_event_bus.emit("source", event)
-
-    # Handler should only be called once (no duplication)
-    assert len(call_counts) == 1
-
-
-def test_decorator_deduplication():
-    """Test that decorator prevents duplicate registrations."""
-    call_counts = []
-
-    # Define the same handler function
-    def handler(source, event):
-        call_counts.append(1)
-
-    # Register using decorator
-    @crewai_event_bus.on(TestEvent)
-    def decorated_handler(source, event):
-        call_counts.append(1)
-
-    # Try to register the same function again using register_handler
-    result = crewai_event_bus.register_handler(
-        TestEvent, cast(Callable[[Any, BaseEvent], None], decorated_handler)
-    )
-
-    # Should fail because it's already registered
-    assert result is False
-
-    # Emit an event
-    event = TestEvent(type="test_event")
-    crewai_event_bus.emit("source", event)
-
-    # Should only be called once
-    assert len(call_counts) == 1
-
-
-def test_handler_unregistration():
-    """Test that handlers can be unregistered."""
-    call_counts = []
-
-    def handler(source, event):
-        call_counts.append(1)
-
-    # Register handler
-    crewai_event_bus.register_handler(TestEvent, handler)
-
-    # Verify it's registered
-    assert crewai_event_bus.get_handler_count(TestEvent) == 1
-
-    # Emit event - should be called
-    event = TestEvent(type="test_event")
-    crewai_event_bus.emit("source", event)
-    assert len(call_counts) == 1
-
-    # Unregister handler
-    result = crewai_event_bus.unregister_handler(TestEvent, handler)
-    assert result is True
-    assert crewai_event_bus.get_handler_count(TestEvent) == 0
-
-    # Emit event again - should not be called
-    crewai_event_bus.emit("source", event)
-    assert len(call_counts) == 1  # Still only 1 call
-
-
-def test_handler_count_tracking():
-    """Test that handler counts are tracked correctly."""
-
-    def handler1(source, event):
-        pass
-
-    def handler2(source, event):
-        pass
-
-    # Initially no handlers
-    assert crewai_event_bus.get_handler_count(TestEvent) == 0
-
-    # Register first handler
-    crewai_event_bus.register_handler(TestEvent, handler1)
-    assert crewai_event_bus.get_handler_count(TestEvent) == 1
-
-    # Register second handler
-    crewai_event_bus.register_handler(TestEvent, handler2)
-    assert crewai_event_bus.get_handler_count(TestEvent) == 2
-
-    # Try to register first handler again (should fail)
-    crewai_event_bus.register_handler(TestEvent, handler1)
-    assert crewai_event_bus.get_handler_count(TestEvent) == 2  # Count unchanged
-
-    # Unregister first handler
-    crewai_event_bus.unregister_handler(TestEvent, handler1)
-    assert crewai_event_bus.get_handler_count(TestEvent) == 1
-
-    # Unregister second handler
-    crewai_event_bus.unregister_handler(TestEvent, handler2)
-    assert crewai_event_bus.get_handler_count(TestEvent) == 0
-
-
-def test_different_event_types_dont_conflict():
-    """Test that handlers for different event types don't interfere."""
-    test_event_calls = []
-    cross_thread_calls = []
-
-    def test_event_handler(source, event):
-        test_event_calls.append(1)
-
-    def cross_thread_handler(source, event):
-        cross_thread_calls.append(1)
-
-    # Register handlers for different event types
-    crewai_event_bus.register_handler(TestEvent, test_event_handler)
-    crewai_event_bus.register_handler(CrossThreadTestEvent, cross_thread_handler)
-
-    # Emit TestEvent
-    test_event = TestEvent(type="test")
-    crewai_event_bus.emit("source", test_event)
-    assert len(test_event_calls) == 1
-    assert len(cross_thread_calls) == 0
-
-    # Emit CrossThreadTestEvent
-    cross_thread_event = CrossThreadTestEvent(type="cross_thread")
-    crewai_event_bus.emit("source", cross_thread_event)
-    assert len(test_event_calls) == 1  # Unchanged
-    assert len(cross_thread_calls) == 1
-
-
-def test_scoped_handlers_with_deduplication():
-    """Test that deduplication works within scoped handlers."""
-    call_counts = []
-
-    def handler(source, event):
-        call_counts.append(1)
-
-    # Register global handler
-    crewai_event_bus.register_handler(TestEvent, handler)
-
-    # Use scoped handlers
-    with crewai_event_bus.scoped_handlers():
-        # Try to register the same handler in scoped context
-        @crewai_event_bus.on(TestEvent)
-        def scoped_handler(source, event):
-            call_counts.append(1)
-
-        # Emit event - should be called by both global and scoped handlers
-        event = TestEvent(type="test_event")
-        crewai_event_bus.emit("source", event)
-
-    # Should have 2 calls (1 global + 1 scoped)
-    assert len(call_counts) == 2
+    assert "Handler 'broken_handler' failed" in out