feat: integrate knowledge store search and background multiplexer with autonomous monitoring into agent pipeline

pr/core/assistant.py

@@ -20,10 +20,16 @@ from pr.tools import (
     search_replace,close_editor,create_diff,apply_patch,
     tail_process, kill_process
 )
+from pr.tools.interactive_control import (
+    start_interactive_session, send_input_to_session, read_session_output,
+    list_active_sessions, close_interactive_session
+)
 from pr.tools.patch import display_file_diff
 from pr.tools.filesystem import display_edit_summary, display_edit_timeline, clear_edit_tracker
 from pr.tools.base import get_tools_definition
 from pr.commands import handle_command
+from pr.core.background_monitor import start_global_monitor, stop_global_monitor, get_global_monitor
+from pr.core.autonomous_interactions import start_global_autonomous, stop_global_autonomous, get_global_autonomous
 
 logger = logging.getLogger('pr')
 logger.setLevel(logging.DEBUG)
@@ -57,6 +63,7 @@ class Assistant:
         self.db_conn = None
         self.autonomous_mode = False
         self.autonomous_iterations = 0
+        self.background_monitoring = False
         self.init_database()
         self.messages.append(init_system_message(args))
 
@@ -69,6 +76,18 @@ class Assistant:
             logger.warning(f"Could not initialize enhanced features: {e}")
             self.enhanced = None
 
+        # Initialize background monitoring components
+        try:
+            start_global_monitor()
+            autonomous = get_global_autonomous()
+            autonomous.start(llm_callback=self._handle_background_updates)
+            self.background_monitoring = True
+            if self.debug:
+                logger.debug("Background monitoring initialized")
+        except Exception as e:
+            logger.warning(f"Could not initialize background monitoring: {e}")
+            self.background_monitoring = False
+
     def init_database(self):
         try:
             logger.debug(f"Initializing database at {DB_PATH}")
@@ -89,6 +108,74 @@ class Assistant:
             logger.error(f"Database initialization error: {e}")
             self.db_conn = None
 
+    def _handle_background_updates(self, updates):
+        """Handle background session updates by injecting them into the conversation."""
+        if not updates or not updates.get('sessions'):
+            return
+
+        # Format the update as a system message
+        update_message = self._format_background_update_message(updates)
+
+        # Inject into current conversation if we're in an active session
+        if self.messages and len(self.messages) > 0:
+            self.messages.append({
+                "role": "system",
+                "content": f"Background session updates: {update_message}"
+            })
+
+            if self.verbose:
+                print(f"{Colors.CYAN}Background update: {update_message}{Colors.RESET}")
+
+    def _format_background_update_message(self, updates):
+        """Format background updates for LLM consumption."""
+        session_summaries = []
+
+        for session_name, session_info in updates.get('sessions', {}).items():
+            summary = session_info.get('summary', f'Session {session_name}')
+            session_summaries.append(f"{session_name}: {summary}")
+
+        if session_summaries:
+            return "Active background sessions: " + "; ".join(session_summaries)
+        else:
+            return "No active background sessions requiring attention."
+
+    def _check_background_updates(self):
+        """Check for pending background updates and display them."""
+        if not self.background_monitoring:
+            return
+
+        try:
+            monitor = get_global_monitor()
+            events = monitor.get_pending_events()
+
+            if events:
+                print(f"\n{Colors.CYAN}Background Events:{Colors.RESET}")
+                for event in events:
+                    event_type = event.get('type', 'unknown')
+                    session_name = event.get('session_name', 'unknown')
+
+                    if event_type == 'session_started':
+                        print(f"  {Colors.GREEN}✓{Colors.RESET} Session '{session_name}' started")
+                    elif event_type == 'session_ended':
+                        print(f"  {Colors.YELLOW}✗{Colors.RESET} Session '{session_name}' ended")
+                    elif event_type == 'output_received':
+                        lines = len(event.get('new_output', {}).get('stdout', []))
+                        print(f"  {Colors.BLUE}📝{Colors.RESET} Session '{session_name}' produced {lines} lines of output")
+                    elif event_type == 'possible_input_needed':
+                        print(f"  {Colors.RED}❓{Colors.RESET} Session '{session_name}' may need input")
+                    elif event_type == 'high_output_volume':
+                        total = event.get('total_lines', 0)
+                        print(f"  {Colors.YELLOW}📊{Colors.RESET} Session '{session_name}' has high output volume ({total} lines)")
+                    elif event_type == 'inactive_session':
+                        inactive_time = event.get('inactive_seconds', 0)
+                        print(f"  {Colors.GRAY}⏰{Colors.RESET} Session '{session_name}' inactive for {inactive_time:.0f}s")
+
+                print()  # Add blank line after events
+
+        except Exception as e:
+            if self.debug:
+                print(f"{Colors.RED}Error checking background updates: {e}{Colors.RESET}")
+
     def execute_tool_calls(self, tool_calls):
         results = []
 
@@ -107,7 +194,10 @@ class Assistant:
                     'run_command': lambda **kw: run_command(**kw),
                     'tail_process': lambda **kw: tail_process(**kw),
                     'kill_process': lambda **kw: kill_process(**kw),
-                    'run_command_interactive': lambda **kw: run_command_interactive(**kw),
+                    'start_interactive_session': lambda **kw: start_interactive_session(**kw),
+                    'send_input_to_session': lambda **kw: send_input_to_session(**kw),
+                    'read_session_output': lambda **kw: read_session_output(**kw),
+                    'close_interactive_session': lambda **kw: close_interactive_session(**kw),
                     'read_file': lambda **kw: read_file(**kw, db_conn=self.db_conn),
                     'write_file': lambda **kw: write_file(**kw, db_conn=self.db_conn),
                     'list_directory': lambda **kw: list_directory(**kw),
@@ -133,6 +223,11 @@ class Assistant:
                     'display_edit_summary': lambda **kw: display_edit_summary(),
                     'display_edit_timeline': lambda **kw: display_edit_timeline(**kw),
                     'clear_edit_tracker': lambda **kw: clear_edit_tracker(),
+                    'start_interactive_session': lambda **kw: start_interactive_session(**kw),
+                    'send_input_to_session': lambda **kw: send_input_to_session(**kw),
+                    'read_session_output': lambda **kw: read_session_output(**kw),
+                    'list_active_sessions': lambda **kw: list_active_sessions(**kw),
+                    'close_interactive_session': lambda **kw: close_interactive_session(**kw),
                     'create_agent': lambda **kw: create_agent(**kw),
                     'list_agents': lambda **kw: list_agents(**kw),
                     'execute_agent_task': lambda **kw: execute_agent_task(**kw),
@@ -264,7 +359,24 @@ class Assistant:
 
         while True:
             try:
-                user_input = input(f"{Colors.BLUE}You>{Colors.RESET} ").strip()
+                # Check for background updates before prompting user
+                if self.background_monitoring:
+                    self._check_background_updates()
+
+                # Create prompt with background status
+                prompt = f"{Colors.BLUE}You"
+                if self.background_monitoring:
+                    try:
+                        from pr.multiplexer import get_all_sessions
+                        sessions = get_all_sessions()
+                        active_count = sum(1 for s in sessions.values() if s.get('status') == 'running')
+                        if active_count > 0:
+                            prompt += f"[{active_count}bg]"
+                    except:
+                        pass
+                prompt += f">{Colors.RESET} "
+
+                user_input = input(prompt).strip()
 
                 if not user_input:
                     continue
@@ -302,6 +414,14 @@ class Assistant:
             except Exception as e:
                 logger.error(f"Error cleaning up enhanced features: {e}")
 
+        # Stop background monitoring
+        if self.background_monitoring:
+            try:
+                stop_global_autonomous()
+                stop_global_monitor()
+            except Exception as e:
+                logger.error(f"Error stopping background monitoring: {e}")
+
         try:
             from pr.multiplexer import cleanup_all_multiplexers
             cleanup_all_multiplexers()

pr/core/autonomous_interactions.py

@@ -0,0 +1,189 @@
+import time
+import threading
+from pr.core.background_monitor import get_global_monitor
+from pr.tools.interactive_control import list_active_sessions, get_session_status, read_session_output
+
+class AutonomousInteractions:
+    def __init__(self, interaction_interval=10.0):
+        self.interaction_interval = interaction_interval
+        self.active = False
+        self.interaction_thread = None
+        self.llm_callback = None
+        self.last_check_time = 0
+
+    def start(self, llm_callback=None):
+        """Start the autonomous interaction loop."""
+        self.llm_callback = llm_callback
+        if self.interaction_thread is None:
+            self.active = True
+            self.interaction_thread = threading.Thread(target=self._interaction_loop, daemon=True)
+            self.interaction_thread.start()
+
+    def stop(self):
+        """Stop the autonomous interaction loop."""
+        self.active = False
+        if self.interaction_thread:
+            self.interaction_thread.join(timeout=2)
+
+    def _interaction_loop(self):
+        """Main loop for autonomous interactions with background processes."""
+        while self.active:
+            try:
+                current_time = time.time()
+                if current_time - self.last_check_time >= self.interaction_interval:
+                    self._check_sessions_and_notify()
+                    self.last_check_time = current_time
+
+                time.sleep(1)  # Check every second for shutdown
+
+            except Exception as e:
+                print(f"Error in autonomous interaction loop: {e}")
+                time.sleep(self.interaction_interval)
+
+    def _check_sessions_and_notify(self):
+        """Check active sessions and determine if LLM notification is needed."""
+        try:
+            sessions = list_active_sessions()
+
+            if not sessions:
+                return  # No active sessions
+
+            sessions_needing_attention = self._identify_sessions_needing_attention(sessions)
+
+            if sessions_needing_attention and self.llm_callback:
+                # Format session updates for LLM
+                updates = self._format_session_updates(sessions_needing_attention)
+                self.llm_callback(updates)
+
+        except Exception as e:
+            print(f"Error checking sessions: {e}")
+
+    def _identify_sessions_needing_attention(self, sessions):
+        """Identify which sessions need LLM attention based on various criteria."""
+        needing_attention = []
+
+        for session_name, session_data in sessions.items():
+            metadata = session_data['metadata']
+            output_summary = session_data['output_summary']
+
+            # Criteria for needing attention:
+
+            # 1. Recent output activity
+            time_since_activity = time.time() - metadata.get('last_activity', 0)
+            if time_since_activity < 30:  # Activity in last 30 seconds
+                needing_attention.append(session_name)
+                continue
+
+            # 2. High output volume (potential completion or error)
+            total_lines = output_summary['stdout_lines'] + output_summary['stderr_lines']
+            if total_lines > 50:  # Arbitrary threshold
+                needing_attention.append(session_name)
+                continue
+
+            # 3. Long-running sessions that might need intervention
+            session_age = time.time() - metadata.get('start_time', 0)
+            if session_age > 300 and time_since_activity > 60:  # 5+ minutes old, inactive for 1+ minute
+                needing_attention.append(session_name)
+                continue
+
+            # 4. Sessions that appear to be waiting for input
+            if self._session_looks_stuck(session_name, session_data):
+                needing_attention.append(session_name)
+                continue
+
+        return needing_attention
+
+    def _session_looks_stuck(self, session_name, session_data):
+        """Determine if a session appears to be stuck waiting for input."""
+        metadata = session_data['metadata']
+
+        # Check if process is still running
+        status = get_session_status(session_name)
+        if not status or not status.get('is_active', False):
+            return False
+
+        time_since_activity = time.time() - metadata.get('last_activity', 0)
+        interaction_count = metadata.get('interaction_count', 0)
+
+        # If running for a while but no interactions, might be waiting
+        session_age = time.time() - metadata.get('start_time', 0)
+        if session_age > 60 and interaction_count == 0 and time_since_activity > 30:
+            return True
+
+        # If had interactions but been quiet for a while
+        if interaction_count > 0 and time_since_activity > 120:  # 2 minutes
+            return True
+
+        return False
+
+    def _format_session_updates(self, session_names):
+        """Format session information for LLM consumption."""
+        updates = {
+            'type': 'background_session_updates',
+            'timestamp': time.time(),
+            'sessions': {}
+        }
+
+        for session_name in session_names:
+            status = get_session_status(session_name)
+            if status:
+                # Get recent output (last 20 lines)
+                try:
+                    recent_output = read_session_output(session_name, lines=20)
+                except:
+                    recent_output = {'stdout': '', 'stderr': ''}
+
+                updates['sessions'][session_name] = {
+                    'status': status,
+                    'recent_output': recent_output,
+                    'summary': self._create_session_summary(status, recent_output)
+                }
+
+        return updates
+
+    def _create_session_summary(self, status, recent_output):
+        """Create a human-readable summary of session status."""
+        summary_parts = []
+
+        process_type = status.get('metadata', {}).get('process_type', 'unknown')
+        summary_parts.append(f"Type: {process_type}")
+
+        is_active = status.get('is_active', False)
+        summary_parts.append(f"Status: {'Active' if is_active else 'Inactive'}")
+
+        if is_active and 'pid' in status:
+            summary_parts.append(f"PID: {status['pid']}")
+
+        age = time.time() - status.get('metadata', {}).get('start_time', 0)
+        summary_parts.append(f"Age: {age:.1f}s")
+
+        output_lines = len(recent_output.get('stdout', '').split('\n')) + len(recent_output.get('stderr', '').split('\n'))
+        summary_parts.append(f"Recent output: {output_lines} lines")
+
+        interaction_count = status.get('metadata', {}).get('interaction_count', 0)
+        summary_parts.append(f"Interactions: {interaction_count}")
+
+        return " | ".join(summary_parts)
+
+# Global autonomous interactions instance
+_global_autonomous = None
+
+def get_global_autonomous():
+    """Get the global autonomous interactions instance."""
+    global _global_autonomous
+    return _global_autonomous
+
+def start_global_autonomous(llm_callback=None):
+    """Start global autonomous interactions."""
+    global _global_autonomous
+    if _global_autonomous is None:
+        _global_autonomous = AutonomousInteractions()
+    _global_autonomous.start(llm_callback)
+    return _global_autonomous
+
+def stop_global_autonomous():
+    """Stop global autonomous interactions."""
+    global _global_autonomous
+    if _global_autonomous:
+        _global_autonomous.stop()
+        _global_autonomous = None

pr/core/background_monitor.py

@@ -0,0 +1,236 @@
+import threading
+import time
+import queue
+from pr.multiplexer import get_all_multiplexer_states, get_multiplexer
+from pr.tools.interactive_control import get_session_status
+
+class BackgroundMonitor:
+    def __init__(self, check_interval=5.0):
+        self.check_interval = check_interval
+        self.active = False
+        self.monitor_thread = None
+        self.event_queue = queue.Queue()
+        self.last_states = {}
+        self.event_callbacks = []
+
+    def start(self):
+        """Start the background monitoring thread."""
+        if self.monitor_thread is None:
+            self.active = True
+            self.monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True)
+            self.monitor_thread.start()
+
+    def stop(self):
+        """Stop the background monitoring thread."""
+        self.active = False
+        if self.monitor_thread:
+            self.monitor_thread.join(timeout=2)
+
+    def add_event_callback(self, callback):
+        """Add a callback function to be called when events are detected."""
+        self.event_callbacks.append(callback)
+
+    def remove_event_callback(self, callback):
+        """Remove an event callback."""
+        if callback in self.event_callbacks:
+            self.event_callbacks.remove(callback)
+
+    def get_pending_events(self):
+        """Get all pending events from the queue."""
+        events = []
+        while not self.event_queue.empty():
+            try:
+                events.append(self.event_queue.get_nowait())
+            except queue.Empty:
+                break
+        return events
+
+    def _monitor_loop(self):
+        """Main monitoring loop that checks for multiplexer activity."""
+        while self.active:
+            try:
+                current_states = get_all_multiplexer_states()
+
+                # Detect changes and events
+                events = self._detect_events(self.last_states, current_states)
+
+                # Queue events for processing
+                for event in events:
+                    self.event_queue.put(event)
+                    # Also call callbacks immediately
+                    for callback in self.event_callbacks:
+                        try:
+                            callback(event)
+                        except Exception as e:
+                            print(f"Error in event callback: {e}")
+
+                self.last_states = current_states.copy()
+                time.sleep(self.check_interval)
+
+            except Exception as e:
+                print(f"Error in background monitor loop: {e}")
+                time.sleep(self.check_interval)
+
+    def _detect_events(self, old_states, new_states):
+        """Detect significant events in multiplexer states."""
+        events = []
+
+        # Check for new sessions
+        for session_name in new_states:
+            if session_name not in old_states:
+                events.append({
+                    'type': 'session_started',
+                    'session_name': session_name,
+                    'metadata': new_states[session_name]['metadata']
+                })
+
+        # Check for ended sessions
+        for session_name in old_states:
+            if session_name not in new_states:
+                events.append({
+                    'type': 'session_ended',
+                    'session_name': session_name
+                })
+
+        # Check for activity in existing sessions
+        for session_name, new_state in new_states.items():
+            if session_name in old_states:
+                old_state = old_states[session_name]
+
+                # Check for output changes
+                old_stdout_lines = old_state['output_summary']['stdout_lines']
+                new_stdout_lines = new_state['output_summary']['stdout_lines']
+                old_stderr_lines = old_state['output_summary']['stderr_lines']
+                new_stderr_lines = new_state['output_summary']['stderr_lines']
+
+                if new_stdout_lines > old_stdout_lines or new_stderr_lines > old_stderr_lines:
+                    # Get the new output
+                    mux = get_multiplexer(session_name)
+                    if mux:
+                        all_output = mux.get_all_output()
+                        new_output = {
+                            'stdout': all_output['stdout'].split('\n')[old_stdout_lines:],
+                            'stderr': all_output['stderr'].split('\n')[old_stderr_lines:]
+                        }
+
+                        events.append({
+                            'type': 'output_received',
+                            'session_name': session_name,
+                            'new_output': new_output,
+                            'total_lines': {
+                                'stdout': new_stdout_lines,
+                                'stderr': new_stderr_lines
+                            }
+                        })
+
+                # Check for state changes
+                old_metadata = old_state['metadata']
+                new_metadata = new_state['metadata']
+
+                if old_metadata.get('state') != new_metadata.get('state'):
+                    events.append({
+                        'type': 'state_changed',
+                        'session_name': session_name,
+                        'old_state': old_metadata.get('state'),
+                        'new_state': new_metadata.get('state')
+                    })
+
+                # Check for process type identification
+                if (old_metadata.get('process_type') == 'unknown' and
+                    new_metadata.get('process_type') != 'unknown'):
+                    events.append({
+                        'type': 'process_identified',
+                        'session_name': session_name,
+                        'process_type': new_metadata.get('process_type')
+                    })
+
+        # Check for sessions needing attention (based on heuristics)
+        for session_name, state in new_states.items():
+            metadata = state['metadata']
+            output_summary = state['output_summary']
+
+            # Heuristic: High output volume might indicate completion or error
+            total_lines = output_summary['stdout_lines'] + output_summary['stderr_lines']
+            if total_lines > 100:  # Arbitrary threshold
+                events.append({
+                    'type': 'high_output_volume',
+                    'session_name': session_name,
+                    'total_lines': total_lines
+                })
+
+            # Heuristic: Long-running session without recent activity
+            time_since_activity = time.time() - metadata.get('last_activity', 0)
+            if time_since_activity > 300:  # 5 minutes
+                events.append({
+                    'type': 'inactive_session',
+                    'session_name': session_name,
+                    'inactive_seconds': time_since_activity
+                })
+
+            # Heuristic: Sessions that might be waiting for input
+            # This would be enhanced with prompt detection in later phases
+            if self._might_be_waiting_for_input(session_name, state):
+                events.append({
+                    'type': 'possible_input_needed',
+                    'session_name': session_name
+                })
+
+        return events
+
+    def _might_be_waiting_for_input(self, session_name, state):
+        """Heuristic to detect if a session might be waiting for input."""
+        metadata = state['metadata']
+        process_type = metadata.get('process_type', 'unknown')
+
+        # Simple heuristics based on process type and recent activity
+        time_since_activity = time.time() - metadata.get('last_activity', 0)
+
+        # If it's been more than 10 seconds since last activity, might be waiting
+        if time_since_activity > 10:
+            return True
+
+        return False
+
+# Global monitor instance
+_global_monitor = None
+
+def get_global_monitor():
+    """Get the global background monitor instance."""
+    global _global_monitor
+    if _global_monitor is None:
+        _global_monitor = BackgroundMonitor()
+    return _global_monitor
+
+def start_global_monitor():
+    """Start the global background monitor."""
+    monitor = get_global_monitor()
+    monitor.start()
+
+def stop_global_monitor():
+    """Stop the global background monitor."""
+    global _global_monitor
+    if _global_monitor:
+        _global_monitor.stop()
+
+# Global monitor instance
+_global_monitor = None
+
+def start_global_monitor():
+    """Start the global background monitor."""
+    global _global_monitor
+    if _global_monitor is None:
+        _global_monitor = BackgroundMonitor()
+        _global_monitor.start()
+    return _global_monitor
+
+def stop_global_monitor():
+    """Stop the global background monitor."""
+    global _global_monitor
+    if _global_monitor:
+        _global_monitor.stop()
+        _global_monitor = None
+
+def get_global_monitor():
+    """Get the global background monitor instance."""
+    global _global_monitor
+    return _global_monitor