merge: resolve conflicts integrating streaming with durability

- Preserve streaming functionality with DraftStreamer integration
- Maintain message journaling and run control from main
- Combine best features from both branches:
  * Progressive streaming with proper error handling
  * Durable message processing with crash recovery
  * Media handling in streaming mode
  * Message hooks integration
  * Placeholder cleanup on errors and cancellation

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

Author: praisonai-triage-agent[bot]

AGENTS.md

@@ -0,0 +1,10 @@
+# Agent Instructions
+
+You are working on the PraisonAI project.
+
+## Project Guidelines
+- Follow the existing code style and conventions
+- Be concise and helpful in responses  
+- Test implementation thoroughly
+- Ensure backward compatibility with existing APIs
+- Follow protocol-driven design: core protocols in praisonaiagents/, heavy implementations in praisonai/

docker/Dockerfile.chat

@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     praisonai_tools \
-    "praisonai>=4.6.56" \
+    "praisonai>=4.6.57" \
     "praisonai[chat]" \
     "embedchain[github,youtube]"
 

docker/Dockerfile.dev

@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     praisonai_tools \
-    "praisonai>=4.6.56" \
+    "praisonai>=4.6.57" \
     "praisonai[ui]" \
     "praisonai[chat]" \
     "praisonai[realtime]" \

docker/Dockerfile.ui

@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     praisonai_tools \
-    "praisonai>=4.6.56" \
+    "praisonai>=4.6.57" \
     "praisonai[ui]" \
     "praisonai[crewai]"
 

examples/python/bot_run_control_example.py

@@ -0,0 +1,238 @@
+#!/usr/bin/env python3
+"""
+Example: Bot In-Flight Run Control
+
+Demonstrates how to use SessionRunControl to provide better UX for bots
+during long-running agent operations. Shows busy feedback, pending message
+handling, and /stop command support.
+
+This example shows the "before" and "after" behavior described in issue #1914.
+"""
+
+import asyncio
+import time
+from typing import Optional
+
+# Mock agent for demonstration
+class MockAgent:
+    """Simple mock agent that simulates long-running tasks."""
+    
+    def __init__(self, name="assistant"):
+        self.name = name
+        self.chat_history = []
+        self._interrupt_controller = None
+    
+    def chat(self, prompt: str) -> str:
+        """Simulate a long-running chat response."""
+        print(f"[Agent] Starting to process: '{prompt[:50]}...'")
+        
+        # Simulate long processing with interrupt checking
+        for i in range(30):  # 3 second task (30 x 0.1s)
+            time.sleep(0.1)
+            
+            # Check for interruption
+            if (self._interrupt_controller and 
+                hasattr(self._interrupt_controller, 'is_set') and
+                self._interrupt_controller.is_set()):
+                reason = getattr(self._interrupt_controller, 'reason', 'unknown')
+                print(f"[Agent] ⚠️ Interrupted: {reason}")
+                raise InterruptedError(f"Task cancelled: {reason}")
+                
+        response = f"Completed analysis of: {prompt}"
+        print(f"[Agent] ✅ Finished: {response}")
+        return response
+
+
+async def demo_without_run_control():
+    """Demonstrate the old behavior (silent blocking)."""
+    print("\n" + "="*60)
+    print("🔴 BEFORE: Without Run Control (Silent Blocking)")
+    print("="*60)
+    
+    from praisonai.bots._session import BotSessionManager
+    
+    # Standard session manager (no run control)
+    session_mgr = BotSessionManager()
+    agent = MockAgent("research-agent")
+    user_id = "user123"
+    
+    print("\n1. User sends long task:")
+    print("   User: 'research quantum computing trends'")
+    
+    # Start first task (this will run for 3 seconds)
+    task1 = asyncio.create_task(
+        session_mgr.chat(agent, user_id, "research quantum computing trends")
+    )
+    
+    # Simulate user sending follow-up messages during execution
+    await asyncio.sleep(0.5)
+    print("\n2. User sends follow-up (this will block silently):")
+    print("   User: 'actually focus on business impact'")
+    
+    task2 = asyncio.create_task(
+        session_mgr.chat(agent, user_id, "actually focus on business impact")  
+    )
+    
+    await asyncio.sleep(0.5)
+    print("\n3. User tries to send /stop (this will also block):")
+    print("   User: '/stop'")
+    
+    task3 = asyncio.create_task(
+        session_mgr.chat(agent, user_id, "/stop")
+    )
+    
+    print("\n⏳ Waiting for all messages to complete...")
+    results = await asyncio.gather(task1, task2, task3)
+    
+    print("\n📊 Results (all processed sequentially with no feedback):")
+    for i, result in enumerate(results, 1):
+        print(f"   {i}. {result}")
+    
+    print("\n❌ Problems with this approach:")
+    print("   - No immediate feedback on follow-up messages")
+    print("   - No way to cancel long-running tasks") 
+    print("   - User doesn't know if bot is working or broken")
+
+
+async def demo_with_run_control():
+    """Demonstrate the new behavior with run control."""
+    print("\n" + "="*60)
+    print("🟢 AFTER: With Run Control (Better UX)")  
+    print("="*60)
+    
+    from praisonai.bots._run_control import SessionRunControl
+    from praisonai.bots._session import BotSessionManager
+    from praisonai.bots._commands import handle_stop_command
+    
+    # Session manager with run control
+    run_control = SessionRunControl(
+        busy_mode="queue",
+        busy_ack_template="⏳ {action} — will process after current task finishes"
+    )
+    session_mgr = BotSessionManager(run_control=run_control)
+    agent = MockAgent("research-agent")
+    user_id = "user123"
+    
+    print("\n1. User sends long task:")
+    print("   User: 'research quantum computing trends'")
+    
+    # Start first task
+    task1 = asyncio.create_task(
+        session_mgr.chat_with_run_control(agent, user_id, "research quantum computing trends")
+    )
+    
+    # Immediate follow-up messages get acknowledgments
+    await asyncio.sleep(0.5)
+    print("\n2. User sends follow-up (gets immediate feedback):")
+    print("   User: 'actually focus on business impact'")
+    
+    result2 = await session_mgr.chat_with_run_control(
+        agent, user_id, "actually focus on business impact"
+    )
+    print(f"   Bot: {result2['response']}")
+    print(f"   Metadata: {result2['metadata']}")
+    
+    await asyncio.sleep(0.5)
+    print("\n3. User decides to stop current task:")
+    print("   User: '/stop'")
+    
+    stop_response = await handle_stop_command(user_id, run_control)
+    print(f"   Bot: {stop_response}")
+    
+    # Wait for first task to complete (it should be cancelled)
+    try:
+        result1 = await task1
+        print(f"\n📊 Task 1 result: {result1['response']}")
+        print(f"   Metadata: {result1['metadata']}")
+    except asyncio.CancelledError:
+        print("\n📊 Task 1 was cancelled as expected")
+    
+    print("\n4. User sends fresh request after stopping:")
+    print("   User: 'what's the weather like?'")
+    
+    result4 = await session_mgr.chat_with_run_control(
+        agent, user_id, "what's the weather like?"
+    )
+    print(f"   Bot: {result4['response']}")
+    
+    print("\n✅ Benefits of this approach:")
+    print("   - Immediate feedback on all messages")
+    print("   - Can cancel long-running tasks with /stop")
+    print("   - Pending messages are queued and processed in order")
+    print("   - User always knows what's happening")
+
+
+async def demo_interrupt_mode():
+    """Demonstrate interrupt mode."""
+    print("\n" + "="*60)
+    print("⚡ BONUS: Interrupt Mode (Cancel and Restart)")
+    print("="*60)
+    
+    from praisonai.bots._run_control import SessionRunControl
+    from praisonai.bots._session import BotSessionManager
+    
+    # Use interrupt mode instead of queue mode
+    run_control = SessionRunControl(busy_mode="interrupt")
+    session_mgr = BotSessionManager(run_control=run_control)
+    agent = MockAgent("research-agent")
+    user_id = "user123"
+    
+    print("\n1. User starts long task:")
+    print("   User: 'research renewable energy for 30 pages'")
+    
+    task1 = asyncio.create_task(
+        session_mgr.chat_with_run_control(agent, user_id, "research renewable energy for 30 pages")
+    )
+    
+    await asyncio.sleep(0.5)
+    print("\n2. User changes mind (interrupts and restarts):")
+    print("   User: 'actually just summarize solar panel efficiency'")
+    
+    result2 = await session_mgr.chat_with_run_control(
+        agent, user_id, "actually just summarize solar panel efficiency"
+    )
+    print(f"   Bot: {result2['response']}")
+    
+    # Wait for original task (should be interrupted)
+    try:
+        result1 = await task1
+        print(f"\n📊 Original task result: {result1['response']}")
+        if result1['metadata'].get('interrupted'):
+            print("   ✅ Original task was properly interrupted")
+    except:
+        print("\n📊 Original task was cancelled")
+    
+    print("\n✅ Interrupt mode is great for:")
+    print("   - When users frequently change direction")
+    print("   - Interactive sessions where latest intent matters most")
+    print("   - Real-time collaboration scenarios")
+
+
+async def main():
+    """Run all demonstrations."""
+    print("🚀 Bot In-Flight Run Control Demo")
+    print("Solving the silent lock problem described in issue #1914")
+    
+    # Show the problem
+    await demo_without_run_control()
+    
+    # Show the solution  
+    await demo_with_run_control()
+    
+    # Show advanced feature
+    await demo_interrupt_mode()
+    
+    print("\n" + "="*60)
+    print("🎯 Summary")
+    print("="*60)
+    print("SessionRunControl solves the silent blocking problem by providing:")
+    print("1. Immediate feedback for mid-run messages")
+    print("2. /stop command support via InterruptController")
+    print("3. Configurable policies (queue/interrupt/steer)")
+    print("4. Run generation tracking to prevent race conditions")
+    print("5. Pending message merging and ordering")
+    print("\nThis makes bot interactions feel responsive and predictable!")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/praisonai-agents/praisonaiagents/bots/config.py

@@ -91,6 +91,16 @@ class BotConfig:
     # When set, stale sessions older than this are auto-reaped.
     session_ttl: int = 0
 
+    # In-flight run control settings for mid-run message handling
+    # busy_mode: Policy for handling messages during active runs
+    #   - "queue": Queue message for next turn (default)
+    #   - "interrupt": Cancel current run and start new one  
+    #   - "steer": Inject message into current run via steering
+    busy_mode: str = "queue"
+    
+    # Template for busy acknowledgment messages (use {action} placeholder)
+    busy_ack: str = "⏳ {action} — will be considered next"
+    
     # Workspace settings for file operation containment and security
     workspace_dir: Optional[str] = None  # default: ~/.praisonai/workspaces/<scope>/<session_key>
     workspace_access: str = "rw"  # "rw" (read-write) | "ro" (read-only) | "none" (copy-on-write sandbox)
@@ -142,6 +152,8 @@ def to_dict(self) -> Dict[str, Any]:
             "ack_emoji": self.ack_emoji,
             "done_emoji": self.done_emoji,
             "session_ttl": self.session_ttl,
+            "busy_mode": self.busy_mode,
+            "busy_ack": self.busy_ack,
             "workspace_dir": self.workspace_dir,
             "workspace_access": self.workspace_access,
             "workspace_scope": self.workspace_scope,

src/praisonai-agents/praisonaiagents/memory/file_memory.py

@@ -115,6 +115,16 @@ class FileMemory:
         "importance_threshold": 0.7,   # Min importance for long-term
         "auto_promote": True,          # Auto-promote important short-term to long-term
     }
+
+    @staticmethod
+    def _sanitise_user_id(user_id: str) -> str:
+        """Reject path traversal in user_id before using it as a directory name."""
+        if not user_id or not isinstance(user_id, str):
+            return "default"
+        if ".." in user_id or "/" in user_id or "\\" in user_id:
+            raise ValueError("user_id must not contain path separators or parent references")
+        safe = user_id.strip()
+        return safe or "default"
 
     def __init__(
         self,
@@ -132,7 +142,7 @@ def __init__(
             config: Configuration overrides
             verbose: Verbosity level (0=quiet, 1=info, 2+=debug)
         """
-        self.user_id = user_id
+        self.user_id = self._sanitise_user_id(user_id)
         self.verbose = verbose
 
         # Set up paths

src/praisonai-agents/praisonaiagents/server/server.py

@@ -117,6 +117,16 @@ def __init__(
         self._server_thread: Optional[threading.Thread] = None
         self._app = None
         self._server = None
+
+    def _authorise_request(self, request) -> bool:
+        """Verify bearer token when auth_token is configured."""
+        token = self.config.auth_token
+        if not token:
+            return True
+        auth = request.headers.get("Authorization", "")
+        if auth.startswith("Bearer ") and auth[7:] == token:
+            return True
+        return request.headers.get("X-Auth-Token") == token
 
     def _create_app(self):
         """Create the ASGI application."""
@@ -137,6 +147,8 @@ async def health(request):
             })
 
         async def events(request):
+            if not self._authorise_request(request):
+                return JSONResponse({"error": "Unauthorized"}, status_code=401)
             client_id = str(uuid.uuid4())
             client = SSEClient(client_id)
             self._clients[client_id] = client
@@ -162,6 +174,8 @@ async def event_generator():
             )
 
         async def publish(request):
+            if not self._authorise_request(request):
+                return JSONResponse({"error": "Unauthorized"}, status_code=401)
             try:
                 data = await request.json()
                 event_type = data.get("type", "message")

src/praisonai-agents/praisonaiagents/tools/ast_grep_tool.py

@@ -31,6 +31,8 @@
 from praisonaiagents._logging import get_logger
 from typing import Optional, List
 
+from ..approval import require_approval
+
 logger = get_logger(__name__)
 
 # Availability cache for performance (checked once per process)
@@ -146,6 +148,7 @@ def ast_grep_search(
         logger.exception("Unexpected error in ast_grep_search")
         return f"Error: {e}"
 
+@require_approval(risk_level="high")
 def ast_grep_rewrite(
     pattern: str,
     replacement: str,

src/praisonai-agents/praisonaiagents/tools/spider_tools.py

@@ -58,9 +58,11 @@ def _ip_blocked(ip: ipaddress._BaseAddress) -> bool:
         pass
 
     try:
-        return _ip_blocked(ipaddress.ip_address(socket.inet_aton(host)))
-    except OSError:
-        pass
+        for info in socket.getaddrinfo(host, None):
+            if _ip_blocked(ipaddress.ip_address(info[4][0])):
+                return True
+    except socket.gaierror:
+        return True
 
     return False