Fix lint errors: remove unused imports from test files

- Remove unused sys and redirect_stderr imports from test_simple_logging_fix.py
- Remove unused sys and pytest imports from tests/test_custom_logger_fix.py
- All lint checks now pass locally

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

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

@@ -19,6 +19,7 @@ class ConsoleFormatter:
     tool_usage_counts: Dict[str, int] = {}
     current_reasoning_branch: Optional[Tree] = None  # Track reasoning status
     _live_paused: bool = False
+    _paused_tree: Optional[Tree] = None
     current_llm_tool_tree: Optional[Tree] = None
 
     def __init__(self, verbose: bool = False):
@@ -120,10 +121,12 @@ class ConsoleFormatter:
         if len(args) == 0 and self._live:
             return
 
-        # Case 3: printing something other than a Tree → terminate live session
+        # Case 3: printing something other than a Tree → temporarily pause live session
         if self._live:
-            self._live.stop()
+            self.pause_live_updates()
-            self._live = None
+            self.console.print(*args, **kwargs)
+            self.resume_live_updates()
+            return
 
         # Finally, pass through to the regular Console.print implementation
         self.console.print(*args, **kwargs)
@@ -132,6 +135,7 @@ class ConsoleFormatter:
         """Pause Live session updates to allow for human input without interference."""
         if not self._live_paused:
             if self._live:
+                self._paused_tree = self._live.renderable
                 self._live.stop()
                 self._live = None
             self._live_paused = True
@@ -139,6 +143,10 @@ class ConsoleFormatter:
     def resume_live_updates(self) -> None:
         """Resume Live session updates after human input is complete."""
         if self._live_paused:
+            if (hasattr(self, '_paused_tree') and self._paused_tree and 
+                hasattr(self._paused_tree, '__rich_console__')):
+                self._live = Live(self._paused_tree, console=self.console, refresh_per_second=4)
+                self._live.start()
             self._live_paused = False
 
     def print_panel(

test_logging_reproduction.py

@@ -0,0 +1,97 @@
+"""
+Reproduction test for issue #3197: Custom logger conflicts with Crew AI logging
+This script demonstrates the problem where custom Python loggers don't work 
+when CrewAI's verbose=True is enabled.
+"""
+import logging
+from crewai import Agent, Task, Crew
+from crewai.tools import BaseTool
+from pydantic import BaseModel, Field
+
+
+class TestInput(BaseModel):
+    message: str = Field(description="Message to log")
+
+
+class CustomLoggingTool(BaseTool):
+    name: str = "custom_logging_tool"
+    description: str = "A tool that uses Python's logging module to demonstrate the conflict"
+    args_schema: type[BaseModel] = TestInput
+
+    def _run(self, message: str) -> str:
+        logger = logging.getLogger("custom_tool_logger")
+        logger.setLevel(logging.INFO)
+        
+        if not logger.handlers:
+            handler = logging.StreamHandler()
+            formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+            handler.setFormatter(formatter)
+            logger.addHandler(handler)
+        
+        logger.info(f"CUSTOM LOGGER MESSAGE: {message}")
+        print(f"PRINT MESSAGE: {message}")
+        
+        return f"Logged message: {message}"
+
+
+def test_logging_with_verbose_true():
+    """Test case that reproduces the logging conflict when verbose=True"""
+    print("=== Testing with verbose=True (should show logging conflict) ===")
+    
+    agent = Agent(
+        role="Test Agent",
+        goal="Test custom logging functionality",
+        backstory="An agent that tests logging",
+        tools=[CustomLoggingTool()],
+        verbose=True
+    )
+    
+    task = Task(
+        description="Use the custom logging tool to log a test message",
+        expected_output="A confirmation that the message was logged",
+        agent=agent
+    )
+    
+    crew = Crew(
+        agents=[agent],
+        tasks=[task],
+        verbose=True
+    )
+    
+    result = crew.kickoff()
+    print(f"Result: {result}")
+
+
+def test_logging_with_verbose_false():
+    """Test case that shows logging works when verbose=False"""
+    print("\n=== Testing with verbose=False (logging should work) ===")
+    
+    agent = Agent(
+        role="Test Agent",
+        goal="Test custom logging functionality", 
+        backstory="An agent that tests logging",
+        tools=[CustomLoggingTool()],
+        verbose=False
+    )
+    
+    task = Task(
+        description="Use the custom logging tool to log a test message",
+        expected_output="A confirmation that the message was logged",
+        agent=agent
+    )
+    
+    crew = Crew(
+        agents=[agent],
+        tasks=[task],
+        verbose=False
+    )
+    
+    result = crew.kickoff()
+    print(f"Result: {result}")
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO)
+    
+    test_logging_with_verbose_false()
+    test_logging_with_verbose_true()

test_simple_logging_fix.py

@@ -0,0 +1,70 @@
+"""
+Simple test to verify the logging fix works without external API calls
+"""
+import logging
+import io
+from contextlib import redirect_stdout
+from crewai.utilities.events.utils.console_formatter import ConsoleFormatter
+from rich.tree import Tree
+
+
+def test_console_formatter_logging_fix():
+    """Test that ConsoleFormatter allows custom logging when Live session is active"""
+    print("Testing ConsoleFormatter logging fix...")
+    
+    logger = logging.getLogger("test_logger")
+    logger.setLevel(logging.INFO)
+    
+    log_buffer = io.StringIO()
+    handler = logging.StreamHandler(log_buffer)
+    formatter = logging.Formatter('CUSTOM_LOG: %(message)s')
+    handler.setFormatter(formatter)
+    logger.addHandler(handler)
+    
+    console_formatter = ConsoleFormatter(verbose=True)
+    
+    tree = Tree("Test Tree")
+    console_formatter.print(tree)
+    
+    assert console_formatter._live is not None, "Live session should be active"
+    
+    logger.info("This should appear in the log buffer")
+    
+    log_output = log_buffer.getvalue()
+    assert "CUSTOM_LOG: This should appear in the log buffer" in log_output, f"Custom log not found in output: {log_output}"
+    
+    assert console_formatter._live is not None, "Live session should still be active after custom logging"
+    
+    print("✅ ConsoleFormatter logging fix test passed!")
+    
+    logger.removeHandler(handler)
+    handler.close()
+
+
+def test_console_formatter_print_behavior():
+    """Test that non-Tree content properly pauses/resumes Live sessions"""
+    print("Testing ConsoleFormatter print behavior...")
+    
+    console_formatter = ConsoleFormatter(verbose=True)
+    
+    tree = Tree("Test Tree")
+    console_formatter.print(tree)
+    
+    assert console_formatter._live is not None, "Live session should be active"
+    
+    stdout_buffer = io.StringIO()
+    with redirect_stdout(stdout_buffer):
+        console_formatter.print("Non-tree content")
+    
+    output = stdout_buffer.getvalue()
+    assert "Non-tree content" in output, f"Non-tree content not found in output: {output}"
+    
+    assert console_formatter._live is not None, "Live session should be restored after printing non-Tree content"
+    
+    print("✅ ConsoleFormatter print behavior test passed!")
+
+
+if __name__ == "__main__":
+    test_console_formatter_logging_fix()
+    test_console_formatter_print_behavior()
+    print("🎉 All simple logging fix tests passed!")

tests/test_custom_logger_fix.py

@@ -0,0 +1,137 @@
+"""
+Tests for issue #3197: Custom logger conflicts with Crew AI logging
+"""
+import logging
+import io
+from unittest.mock import patch
+from crewai import Agent, Task, Crew
+from crewai.tools import BaseTool
+from pydantic import BaseModel, Field
+
+
+class TestInput(BaseModel):
+    message: str = Field(description="Message to log")
+
+
+class CustomLoggingTool(BaseTool):
+    name: str = "custom_logging_tool"
+    description: str = "A tool that uses Python's logging module"
+    args_schema: type[BaseModel] = TestInput
+
+    def _run(self, message: str) -> str:
+        logger = logging.getLogger("test_custom_logger")
+        logger.setLevel(logging.INFO)
+        
+        if not logger.handlers:
+            handler = logging.StreamHandler()
+            formatter = logging.Formatter('CUSTOM_LOG: %(message)s')
+            handler.setFormatter(formatter)
+            logger.addHandler(handler)
+        
+        logger.info(f"Custom logger message: {message}")
+        print(f"Print message: {message}")
+        
+        return f"Logged: {message}"
+
+
+def test_custom_logger_with_verbose_false():
+    """Test that custom loggers work when verbose=False"""
+    agent = Agent(
+        role="Test Agent",
+        goal="Test logging",
+        backstory="Testing agent",
+        tools=[CustomLoggingTool()],
+        verbose=False
+    )
+    
+    task = Task(
+        description="Log a test message",
+        expected_output="Confirmation of logging",
+        agent=agent
+    )
+    
+    crew = Crew(
+        agents=[agent],
+        tasks=[task],
+        verbose=False
+    )
+    
+    with patch('sys.stdout', new_callable=io.StringIO) as mock_stdout:
+        result = crew.kickoff()
+        output = mock_stdout.getvalue()
+        
+        assert "Custom logger message" in output or "Print message" in output
+        assert result is not None
+
+
+def test_custom_logger_with_verbose_true():
+    """Test that custom loggers work when verbose=True after the fix"""
+    agent = Agent(
+        role="Test Agent", 
+        goal="Test logging",
+        backstory="Testing agent",
+        tools=[CustomLoggingTool()],
+        verbose=True
+    )
+    
+    task = Task(
+        description="Log a test message",
+        expected_output="Confirmation of logging", 
+        agent=agent
+    )
+    
+    crew = Crew(
+        agents=[agent],
+        tasks=[task],
+        verbose=True
+    )
+    
+    with patch('sys.stdout', new_callable=io.StringIO) as mock_stdout:
+        result = crew.kickoff()
+        output = mock_stdout.getvalue()
+        
+        assert "Custom logger message" in output or "Print message" in output
+        assert result is not None
+
+
+def test_console_formatter_pause_resume():
+    """Test that ConsoleFormatter properly pauses and resumes Live sessions"""
+    from crewai.utilities.events.utils.console_formatter import ConsoleFormatter
+    from rich.tree import Tree
+    
+    formatter = ConsoleFormatter(verbose=True)
+    
+    tree = Tree("Test Tree")
+    formatter.print(tree)
+    
+    assert formatter._live is not None
+    assert not formatter._live_paused
+    
+    formatter.pause_live_updates()
+    assert formatter._live_paused
+    assert formatter._live is None
+    assert formatter._paused_tree is not None
+    
+    formatter.resume_live_updates()
+    assert not formatter._live_paused
+    assert formatter._live is not None
+
+
+def test_console_formatter_non_tree_printing():
+    """Test that non-Tree content properly pauses/resumes Live sessions"""
+    from crewai.utilities.events.utils.console_formatter import ConsoleFormatter
+    from rich.tree import Tree
+    
+    formatter = ConsoleFormatter(verbose=True)
+    
+    tree = Tree("Test Tree")
+    formatter.print(tree)
+    
+    assert formatter._live is not None
+    
+    with patch('sys.stdout', new_callable=io.StringIO) as mock_stdout:
+        formatter.print("Non-tree content")
+        output = mock_stdout.getvalue()
+        
+        assert "Non-tree content" in output
+        assert formatter._live is not None