fix: address critical bugs in message steering implementation

- Fix INTERRUPT detection in tool_execution.py (was using broken substring check)
- Fix rate limiter blocking urgent messages during tool execution
- Add INTERRUPT priority bypass in message queue (URGENT.value + 1)
- Add proper error handling in execution_mixin.py with graceful degradation
- Extend MessageSteeringProtocol with enabled property for contract compliance
- Move test file to proper tests/integration/ directory structure
- Add mandatory live agentic test with real LLM execution
- Improve prompt formatting with clear separators

Fixes critical bugs identified by Greptile, CodeRabbit reviewers.
All changes maintain backward compatibility.

Co-authored-by: Mervin Praison <MervinPraison@users.noreply.github.com>

Author: praisonai-triage-agent[bot]

src/praisonai-agents/praisonaiagents/agent/execution_mixin.py

@@ -808,13 +808,18 @@ def run_chat():
 
                             # Check for steering messages before starting chat
                             if hasattr(self, '_check_steering_messages'):
-                                steering_msg = self._check_steering_messages()
-                                if steering_msg:
-                                    # Inject steering message into prompt
-                                    prompt_with_steering = f"{prompt}{steering_msg}"
-                                    current_status[0] = "Processing steering guidance..."
-                                    result_holder[0] = self.chat(prompt_with_steering, **kwargs)
-                                else:
+                                try:
+                                    steering_msg = self._check_steering_messages()
+                                    if steering_msg:
+                                        # Inject steering message into prompt with clear separator
+                                        prompt_with_steering = f"{prompt}\n\n{steering_msg}"
+                                        current_status[0] = "Processing steering guidance..."
+                                        result_holder[0] = self.chat(prompt_with_steering, **kwargs)
+                                    else:
+                                        result_holder[0] = self.chat(prompt, **kwargs)
+                                except Exception as e:
+                                    logger.warning(f"Steering check failed, continuing without steering: {e}")
+                                    current_status[0] = "Steering failed, proceeding normally..."
                                     result_holder[0] = self.chat(prompt, **kwargs)
                             else:
                                 result_holder[0] = self.chat(prompt, **kwargs)

src/praisonai-agents/praisonaiagents/agent/message_steering.py

@@ -56,7 +56,11 @@ def queue_message(self, message: str, priority: int = 5) -> str:
         )
 
         # Use existing message queue with priority mapping
-        queue_priority = min(priority, MessagePriority.URGENT.value)
+        # Special handling for INTERRUPT - give it maximum priority
+        if steering_priority == SteeringPriority.INTERRUPT:
+            queue_priority = MessagePriority.URGENT.value + 1  # Higher than URGENT
+        else:
+            queue_priority = min(priority, MessagePriority.URGENT.value)
         success = self._message_queue.enqueue(
             content=steering_msg,
             priority=queue_priority,
@@ -90,7 +94,8 @@ def process_steering(self, context: Optional[Dict[str, Any]] = None) -> bool:
         Process pending steering messages.
         
         This is called during agent execution to check for and process
-        any steering messages. Uses rate limiting to avoid excessive checking.
+        any steering messages. Uses rate limiting to avoid excessive checking,
+        but allows INTERRUPT priority messages to bypass rate limiting.
         
         Args:
             context: Execution context that can be updated with steering info
@@ -101,9 +106,17 @@ def process_steering(self, context: Optional[Dict[str, Any]] = None) -> bool:
         if not self._enabled:
             return False
 
-        # Rate limiting - only check every 100ms
+        # Check if we have high priority messages that bypass rate limiting
+        has_urgent_messages = False
+        all_messages = self._message_queue.get_all()
+        for msg in all_messages:
+            if isinstance(msg, SteeringMessage) and msg.priority.value >= SteeringPriority.HIGH.value:
+                has_urgent_messages = True
+                break
+        
+        # Rate limiting - only check every 100ms unless we have urgent messages
         current_time = time.time()
-        if current_time - self._last_check < self._check_interval:
+        if not has_urgent_messages and current_time - self._last_check < self._check_interval:
             return False
 
         self._last_check = current_time

src/praisonai-agents/praisonaiagents/agent/message_steering_protocols.py

@@ -109,6 +109,16 @@ def has_pending_messages(self) -> bool:
             True if messages are pending, False otherwise
         """
         ...
+    
+    @property
+    def enabled(self) -> bool:
+        """
+        Whether message steering is enabled.
+        
+        Returns:
+            True if steering is enabled, False otherwise
+        """
+        ...
 
 
 @runtime_checkable

src/praisonai-agents/praisonaiagents/agent/tool_execution.py

@@ -197,11 +197,16 @@ def _execute_tool_with_context(self, function_name, arguments, state, tool_call_
         try:
             # Check for steering messages before tool execution
             if hasattr(self, '_check_steering_messages'):
-                steering_msg = self._check_steering_messages()
-                if steering_msg and "INTERRUPT" in steering_msg:
-                    # High priority steering - interrupt tool execution
-                    logger.info(f"Tool {function_name} execution interrupted by steering message")
-                    return f"Tool execution interrupted by user guidance: {steering_msg}"
+                try:
+                    steering_msg = self._check_steering_messages()
+                    if steering_msg:
+                        # Check if steering message indicates interruption priority
+                        # (SteeringMixin formats HIGH/URGENT/INTERRUPT with specific prefixes)
+                        if any(marker in steering_msg for marker in ["[URGENT USER GUIDANCE]", "[INTERRUPT]"]):
+                            logger.info(f"Tool {function_name} execution interrupted by high-priority steering")
+                            return f"Tool execution interrupted by user guidance: {steering_msg}"
+                except Exception as e:
+                    logger.warning(f"Steering check failed, continuing with tool execution: {e}")
 
             # Trigger BEFORE_TOOL hook
             from ..hooks import HookEvent, BeforeToolInput

src/praisonai-agents/tests/integration/test_message_steering.py

@@ -0,0 +1,157 @@
+#!/usr/bin/env python3
+"""
+Integration tests for message steering capability.
+
+Tests the real-time message steering implementation including
+real agentic tests with actual LLM calls.
+"""
+import time
+import threading
+import pytest
+from praisonaiagents import Agent
+
+
+def test_message_steering_basic():
+    """Test basic message steering functionality."""
+    # Create agent with message steering enabled
+    agent = Agent(
+        name="test_agent",
+        instructions="You are a helpful assistant. Acknowledge any user guidance.",
+        message_steering=True,
+        llm="gpt-4o-mini"
+    )
+    
+    # Verify steering is enabled
+    assert agent.message_steering_enabled, "Message steering should be enabled"
+    
+    # Test queueing messages
+    msg_id = agent.steer("Focus on being concise")
+    assert msg_id, "Should return message ID"
+    
+    status = agent.get_steering_status()
+    assert status["enabled"], "Steering should be enabled"
+    assert status["pending_count"] > 0, "Should have pending messages"
+
+
+def test_message_steering_disabled():
+    """Test that steering is disabled by default."""
+    agent = Agent(name="test_agent", instructions="You are helpful")
+    
+    # Verify steering is disabled
+    assert not agent.message_steering_enabled, "Message steering should be disabled by default"
+    
+    # Test steering call returns empty ID
+    msg_id = agent.steer("This should be ignored")
+    assert msg_id == "", "Should return empty string when disabled"
+    
+    status = agent.get_steering_status()
+    assert not status["enabled"], "Steering should be disabled"
+    assert status["pending_count"] == 0, "Should have no pending messages"
+
+
+def test_message_steering_integration():
+    """Test integration with execution (smoke test only - no actual LLM call)."""
+    agent = Agent(
+        name="integration_test",
+        instructions="You are helpful",
+        message_steering=True
+    )
+    
+    # Add a steering message
+    msg_id = agent.steer("Please be very brief", priority=10)
+    assert msg_id, "Should queue message"
+    
+    # Check that steering check method exists
+    assert hasattr(agent, '_check_steering_messages'), "Should have steering check method"
+    
+    # Test the steering check method
+    steering_msg = agent._check_steering_messages()
+    assert steering_msg is not None, "Should return steering message"
+    assert "USER GUIDANCE" in steering_msg, "Should format as guidance"
+    assert "brief" in steering_msg.lower(), "Should contain original message"
+
+
+def test_message_steering_live_execution():
+    """Test steering injection during live LLM execution (MANDATORY real agentic test)."""
+    agent = Agent(
+        name="steering_live_test",
+        instructions="You are a helpful assistant. Always acknowledge user guidance when provided.",
+        message_steering=True,
+        llm="gpt-4o-mini"
+    )
+    
+    # Container to capture agent response
+    result_container = {}
+    
+    def run_agent():
+        """Execute agent in background thread."""
+        try:
+            result = agent.start("Explain quantum computing in detail")
+            result_container["response"] = result
+            result_container["success"] = True
+        except Exception as e:
+            result_container["error"] = str(e)
+            result_container["success"] = False
+    
+    # Start agent execution in background
+    thread = threading.Thread(target=run_agent)
+    thread.start()
+    
+    # Allow execution to start
+    time.sleep(0.5)
+    
+    # Send steering message while agent is running
+    msg_id = agent.steer("Keep your explanation under 100 words and focus on practical applications", priority=10)
+    assert msg_id, "Should queue steering message"
+    
+    # Wait for execution to complete (with timeout)
+    thread.join(timeout=30)
+    
+    # Verify execution completed successfully
+    assert result_container.get("success", False), f"Agent execution failed: {result_container.get('error', 'Unknown error')}"
+    
+    response = result_container.get("response", "")
+    assert response, "Should return non-empty response from LLM"
+    assert len(response) > 10, "Response should be substantial"
+    
+    # Print response for manual verification
+    print(f"\n🤖 Agent Response:\n{response}")
+    
+    # Verify steering status after execution
+    final_status = agent.get_steering_status()
+    # Note: Message may have been processed during execution
+    print(f"\n📊 Final steering status: {final_status}")
+
+
+def test_message_steering_priority_handling():
+    """Test different priority levels for steering messages."""
+    agent = Agent(
+        name="priority_test",
+        instructions="You are helpful",
+        message_steering=True
+    )
+    
+    # Test different priority levels
+    low_msg = agent.steer("Low priority guidance", priority=1)
+    normal_msg = agent.steer("Normal priority guidance", priority=5) 
+    high_msg = agent.steer("High priority guidance", priority=10)
+    urgent_msg = agent.steer("Urgent guidance", priority=20)
+    interrupt_msg = agent.steer("Interrupt guidance", priority=30)
+    
+    # All should return message IDs
+    assert all([low_msg, normal_msg, high_msg, urgent_msg, interrupt_msg])
+    
+    # Check pending count
+    status = agent.get_steering_status()
+    assert status["pending_count"] == 5
+    
+    # Process one message
+    steering_msg = agent._check_steering_messages()
+    # Should process highest priority first
+    assert steering_msg is not None
+    assert "Interrupt" in steering_msg or "Urgent" in steering_msg
+
+
+if __name__ == "__main__":
+    # Run tests manually if executed directly
+    pytest.main([__file__, "-v"])