enh: The optimization eliminates the overhead of:

Forking a new Python process (~1.2ms per fork_exec)
Initializing the Python interpreter
Loading modules and dependencies
Setting up the subprocess communication pipes

Signed-off-by: habeck <habeck@us.ibm.com>

Author: tedhabeck

cpex/framework/isolated/client.py

@@ -223,6 +223,13 @@ async def initialize(self) -> None:
         else:
             logger.info("Using cached venv, skipping requirements installation")
 
+    async def cleanup(self) -> None:
+        """Cleanup resources, including stopping the worker process."""
+        if self.comm:
+            logger.info("Stopping worker process for plugin '%s'", self.name)
+            self.comm.stop_worker()
+            self.comm = None
+
     def _validate_hook_invocation(self, hook_type: str) -> type[PluginResult]:
         """Validate hook type and communication channel.
 

cpex/framework/isolated/venv_comm.py

@@ -11,16 +11,19 @@
 import os
 import subprocess
 import sys
+import threading
+import uuid
 from pathlib import Path
-from typing import Any
+from queue import Empty, Queue
+from typing import Any, Optional
 
 import orjson
 
 logger = logging.getLogger(__name__)
 
 
 class VenvProcessCommunicator:
-    """Handles communication with child processes in different virtual environments."""
+    """Handles communication with a long-running child process in a different virtual environment."""
 
     def __init__(self, venv_path: str) -> None:
         """
@@ -31,6 +34,12 @@ def __init__(self, venv_path: str) -> None:
         """
         self.venv_path = Path(venv_path)
         self.python_executable = self._get_python_executable()
+        self.process: Optional[subprocess.Popen] = None
+        self.reader_thread: Optional[threading.Thread] = None
+        self.stderr_thread: Optional[threading.Thread] = None
+        self.response_queues: dict[str, Queue] = {}
+        self.lock = threading.Lock()
+        self.running = False
         logger.info("cwd: %s", os.getcwd())
 
     def _get_python_executable(self):
@@ -57,47 +66,199 @@ def install_requirements(self, requirements_file: str) -> None:
             if rc != 0:
                 raise Exception(f"Failed to install requirements from {requirements_file}")
 
-    def send_task(self, script_path: str, task_data: Any) -> Any:
+    def start_worker(self, script_path: str) -> None:
         """
-        Send a task to child process and get response.
+        Start the long-running worker process.
 
         Args:
-            script_path (str): Path to the child script
-            task_data (dict): Data to send to child process
-
-        Returns:
-            dict: Response from child process
+            script_path (str): Path to the worker script
         """
-        process = None
+        if self.running:
+            logger.warning("Worker process already running")
+            return
+
         try:
-            # Prepare input data as JSON
-            input_json = orjson.dumps(task_data).decode()
             # Start child process
-            process = subprocess.Popen(
+            self.process = subprocess.Popen(
                 [self.python_executable, script_path],
                 stdin=subprocess.PIPE,
                 stdout=subprocess.PIPE,
                 stderr=subprocess.PIPE,
                 text=True,
-                cwd=os.getcwd(),  # Maintain current working directory
+                bufsize=1,  # Line buffered
+                cwd=os.getcwd(),
             )
 
-            # Send data and get response
-            stdout, stderr = process.communicate(input=input_json, timeout=30)
+            self.running = True
+
+            # Start reader thread to handle responses
+            self.reader_thread = threading.Thread(target=self._read_responses, daemon=True)
+            self.reader_thread.start()
+
+            # Start stderr reader thread to capture errors
+            self.stderr_thread = threading.Thread(target=self._read_stderr, daemon=True)
+            self.stderr_thread.start()
+
+            logger.info("Worker process started with PID: %s", self.process.pid)
+
+        except Exception as e:
+            self.running = False
+            raise RuntimeError(f"Failed to start worker process: {e}")
+
+    def _read_stderr(self) -> None:
+        """Background thread to read and log stderr from worker process."""
+        if not self.process or not self.process.stderr:
+            return
+
+        while self.running and self.process and self.process.stderr:
+            try:
+                line = self.process.stderr.readline()
+                if not line:
+                    break
+                # Log stderr output from worker
+                logger.debug("Worker stderr: %s", line.strip())
+            except Exception as e:
+                logger.error("Error reading stderr: %s", e)
+                break
+
+    def _read_responses(self) -> None:
+        """Background thread to read responses from worker process."""
+        while self.running and self.process and self.process.stdout:
+            try:
+                line = self.process.stdout.readline()
+                if not line:
+                    # Process has terminated
+                    logger.warning("Worker process stdout closed")
+                    break
+
+                line = line.strip()
+                if not line:
+                    # Empty line, skip
+                    continue
+
+                try:
+                    response = json.loads(line)
+                    request_id = response.get("request_id")
+
+                    if request_id:
+                        with self.lock:
+                            if request_id in self.response_queues:
+                                self.response_queues[request_id].put(response)
+                                logger.debug("Response queued for request_id: %s", request_id)
+                            else:
+                                logger.warning("Received response for unknown request_id: %s", request_id)
+                    else:
+                        logger.warning("Received response without request_id: %s", line[:100])
+
+                except json.JSONDecodeError as e:
+                    logger.error("Failed to decode response: %s, line: %s", e, line[:200])
+
+            except Exception as e:
+                logger.exception("Error reading response: %s", e)
+                break
+
+        self.running = False
+        logger.info("Response reader thread terminated")
+
+    def send_task(self, script_path: str, task_data: Any, timeout: float = 30.0) -> Any:
+        """
+        Send a task to the long-running worker process and get response.
+
+        Args:
+            script_path (str): Path to the child script (used for worker initialization)
+            task_data (dict): Data to send to child process
+            timeout (float): Timeout in seconds for waiting for response
+
+        Returns:
+            dict: Response from child process
+        """
+        # Start worker if not running
+        if not self.running:
+            self.start_worker(script_path)
+
+        # Generate unique request ID
+        request_id = str(uuid.uuid4())
+        task_data["request_id"] = request_id
+
+        # Create response queue for this request
+        response_queue: Queue = Queue()
+        with self.lock:
+            self.response_queues[request_id] = response_queue
 
-            if process.returncode != 0:
-                raise RuntimeError(f"Child process failed: {stderr}")
+        try:
+            # Send task to worker
+            input_json = orjson.dumps(task_data).decode()
+            if self.process and self.process.stdin:
+                self.process.stdin.write(input_json + "\n")
+                self.process.stdin.flush()
+            else:
+                raise RuntimeError("Worker process stdin not available")
 
-            # Parse response
+            # Wait for response
             try:
-                response = json.loads(stdout.strip())
+                response = response_queue.get(timeout=timeout)
+
+                # Check for errors in response
+                if response.get("status") == "error":
+                    raise RuntimeError(f"Worker process error: {response.get('message')}")
+
+                # Remove request_id from response before returning
+                response.pop("request_id", None)
                 return response
-            except json.JSONDecodeError:
-                raise RuntimeError(f"Invalid JSON response from child: {stdout}")
 
-        except subprocess.TimeoutExpired:
-            if process:
-                process.kill()
-            raise RuntimeError("Child process timed out")
+            except Empty:
+                raise RuntimeError(f"Worker process timed out after {timeout} seconds")
+
+        finally:
+            # Clean up response queue
+            with self.lock:
+                self.response_queues.pop(request_id, None)
+
+    def stop_worker(self) -> None:
+        """Stop the long-running worker process."""
+        if not self.running:
+            return
+
+        self.running = False
+
+        try:
+            if self.process:
+                # Send shutdown signal
+                if self.process.stdin:
+                    try:
+                        shutdown_task = {"task_type": "shutdown", "request_id": "shutdown"}
+                        self.process.stdin.write(json.dumps(shutdown_task) + "\n")
+                        self.process.stdin.flush()
+                    except Exception as e:
+                        logger.warning("Failed to send shutdown signal: %s", e)
+
+                # Wait for process to terminate gracefully
+                try:
+                    self.process.wait(timeout=5.0)
+                except subprocess.TimeoutExpired:
+                    logger.warning("Worker process did not terminate gracefully, killing it")
+                    self.process.kill()
+                    self.process.wait()
+
+                logger.info("Worker process stopped")
+
         except Exception as e:
-            raise RuntimeError(f"Communication error: {e}")
+            logger.error("Error stopping worker process: %s", e)
+
+        finally:
+            self.process = None
+            if self.reader_thread and self.reader_thread.is_alive():
+                self.reader_thread.join(timeout=2.0)
+            self.reader_thread = None
+            if self.stderr_thread and self.stderr_thread.is_alive():
+                self.stderr_thread.join(timeout=2.0)
+            self.stderr_thread = None
+
+    def is_alive(self) -> bool:
+        """Check if the worker process is alive and running."""
+        return self.running and self.process is not None and self.process.poll() is None
+
+    def __del__(self):
+        """Cleanup when object is destroyed."""
+        if hasattr(self, 'running'):
+            self.stop_worker()

cpex/framework/isolated/worker.py

@@ -100,24 +100,70 @@ async def process_task(task_data):
 
 
 async def main():
-    """Main function - read from stdin, process, write to stdout."""
+    """Main function - continuously read from stdin, process tasks, write to stdout."""
+    logger.info("Worker process started, waiting for tasks...")
+    
     try:
-        # Read input from parent process
-        input_data = sys.stdin.read()
-        task_data = json.loads(input_data)
-
-        # Process the task
-        response = await process_task(task_data)
-        serializable_response = response.model_dump(mode="json") if response else None
-        # Send response back to parent
-        print(json.dumps(serializable_response))
-
-    except json.JSONDecodeError:
-        error_response = {"status": "error", "message": "Invalid JSON input"}
-        print(json.dumps(error_response))
+        # Continuously read and process tasks
+        while True:
+            try:
+                # Read one line at a time
+                line = sys.stdin.readline()
+                
+                # Check for EOF
+                if not line:
+                    logger.info("EOF received, shutting down worker")
+                    break
+                
+                # Parse the task
+                task_data = json.loads(line.strip())
+                request_id = task_data.get("request_id", "unknown")
+                
+                # Check for shutdown signal
+                if task_data.get("task_type") == "shutdown":
+                    logger.info("Shutdown signal received")
+                    response = {"status": "success", "message": "Shutting down", "request_id": request_id}
+                    print(json.dumps(response), flush=True)
+                    break
+                
+                # Process the task
+                response = await process_task(task_data)
+                
+                # Serialize response
+                if response:
+                    serializable_response = response.model_dump(mode="json")
+                else:
+                    serializable_response = {"status": "success"}
+                
+                # Add request_id to response
+                serializable_response["request_id"] = request_id
+                
+                # Send response back to parent (one line per response)
+                print(json.dumps(serializable_response), flush=True)
+                
+            except json.JSONDecodeError as e:
+                error_response = {
+                    "status": "error",
+                    "message": f"Invalid JSON input: {str(e)}",
+                    "request_id": task_data.get("request_id", "unknown") if 'task_data' in locals() else "unknown"
+                }
+                print(json.dumps(error_response), flush=True)
+                
+            except Exception as e:
+                logger.error("Error processing task: %s", str(e))
+                error_response = {
+                    "status": "error",
+                    "message": f"Unexpected error: {str(e)}",
+                    "request_id": task_data.get("request_id", "unknown") if 'task_data' in locals() else "unknown"
+                }
+                print(json.dumps(error_response), flush=True)
+    
+    except KeyboardInterrupt:
+        logger.info("Worker interrupted")
     except Exception as e:
-        error_response = {"status": "error", "message": f"Unexpected error: {str(e)}"}
-        print(json.dumps(error_response))
+        logger.exception("Fatal error in worker main loop")
+    finally:
+        logger.info("Worker process shutting down")
 
 
 if __name__ == "__main__":

tests/unit/cpex/framework/isolated/test_client.py

@@ -650,6 +650,25 @@ async def test_initialize_with_invalid_cache(self, mock_save_metadata, mock_cach
         # Should install requirements when cache is invalid
         mock_comm.install_requirements.assert_called_once()
         mock_save_metadata.assert_called_once()
+    @pytest.mark.asyncio
+    async def test_cleanup(self, plugin):
+        """Test cleanup method stops worker process."""
+        mock_comm = MagicMock()
+        plugin.comm = mock_comm
+        
+        await plugin.cleanup()
+        
+        mock_comm.stop_worker.assert_called_once()
+        assert plugin.comm is None
+
+    @pytest.mark.asyncio
+    async def test_cleanup_no_comm(self, plugin):
+        """Test cleanup when comm is None."""
+        plugin.comm = None
+        
+        # Should not raise error
+        await plugin.cleanup()
+
 
 
 # Made with Bob