Merge branch 'feature/zombie-process-fix' into refactor/shell-executor

Author: tumf

.github/workflows/publish.yml

@@ -1,15 +1,17 @@
 name: Publish
+
 on:
   push:
     tags:
       - "v*"
-    strategy:
-      matrix:
-        python-version: ["3.11"]
 
 jobs:
   test:
     runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.11"]
+
     steps:
       - uses: actions/checkout@v4
 
@@ -23,14 +25,13 @@ jobs:
           python -m pip install --upgrade pip
           pip install uv
 
-      - name: Install dev/test dependencies
+      - name: Install dependencies
         run: |
-          pip install -e ".[dev]"
-          pip install -e ".[test]"
+          uv pip install -e ".[dev,test]"
 
-      - name: Run tests
+      - name: Run tests and checks
         run: |
-          make check
+          make all
 
   publish:
     needs: test
@@ -44,14 +45,13 @@ jobs:
 
       - name: Update version from tag
         run: |
-          # Strip 'v' prefix from tag and update version.py
           VERSION=${GITHUB_REF#refs/tags/v}
           echo "__version__ = \"${VERSION}\"" > src/mcp_shell_server/version.py
 
-      - name: Set up Python ${{ matrix.python-version }}
+      - name: Set up Python 3.11
         uses: actions/setup-python@v5
         with:
-          python-version: ${{ matrix.python-version }}
+          python-version: "3.11"
 
       - name: Install uv
         run: |
@@ -60,10 +60,11 @@ jobs:
 
       - name: Build package
         run: |
-          uv build
+          uv pip install build
+          python -m build
 
       - name: Publish to PyPI
-        env:
-          PYPI_TOKEN: ${{ secrets.PYPI_TOKEN }}
-        run: |
-          uv publish --token $PYPI_TOKEN
+        uses: pypa/gh-action-pypi-publish@release/v1
+        with:
+          password: ${{ secrets.PYPI_TOKEN }}
+          password: ${{ secrets.PYPI_TOKEN }}

.github/workflows/test.yml

@@ -26,18 +26,17 @@ jobs:
           python -m pip install --upgrade pip
           pip install uv
 
-      - name: Install dev/test dependencies
+      - name: Install dependencies
         run: |
-          pip install -e ".[dev]"
-          pip install -e ".[test]"
+          uv pip install -e ".[dev,test]"
 
-      - name: Run lint and typecheck
+      - name: Run format checks and typecheck
         run: |
-          make lint typecheck
+          make check
 
       - name: Run tests with coverage
         run: |
-          pytest --cov --cov-report=xml
+          make coverage
 
       - name: Upload coverage to Codecov
         uses: codecov/codecov-action@v5

pyproject.toml

@@ -41,6 +41,10 @@ asyncio_mode = "strict"
 testpaths = "tests"
 # Set default event loop scope for async tests 
 asyncio_default_fixture_loop_scope = "function"
+markers = [
+    "macos: marks tests that should only run on macOS",
+    "slow: marks tests as slow running",
+]
 filterwarnings = [
     "ignore::RuntimeWarning:selectors:",
     "ignore::pytest.PytestUnhandledCoroutineWarning:",

src/mcp_shell_server/process_manager.py

@@ -3,19 +3,131 @@
 import asyncio
 import logging
 import os
-from typing import IO, Any, Dict, List, Optional, Tuple, Union
+import signal
+from typing import IO, Any, Dict, List, Optional, Set, Tuple, Union
+from weakref import WeakSet
 
 
 class ProcessManager:
     """Manages process creation, execution, and cleanup for shell commands."""
 
+    def __init__(self):
+        """Initialize ProcessManager with signal handling setup."""
+        self._processes: Set[asyncio.subprocess.Process] = WeakSet()
+        self._original_sigint_handler = None
+        self._original_sigterm_handler = None
+        self._setup_signal_handlers()
+
+    def _setup_signal_handlers(self) -> None:
+        """Set up signal handlers for graceful process management."""
+        if os.name != "posix":
+            return
+
+        def handle_termination(signum: int, _: Any) -> None:
+            """Handle termination signals by cleaning up processes."""
+            if self._processes:
+                for process in self._processes:
+                    try:
+                        if process.returncode is None:
+                            process.terminate()
+                    except Exception as e:
+                        logging.warning(
+                            f"Error terminating process on signal {signum}: {e}"
+                        )
+
+            # Restore original handler and re-raise signal
+            if signum == signal.SIGINT and self._original_sigint_handler:
+                signal.signal(signal.SIGINT, self._original_sigint_handler)
+            elif signum == signal.SIGTERM and self._original_sigterm_handler:
+                signal.signal(signal.SIGTERM, self._original_sigterm_handler)
+
+            # Re-raise signal
+            os.kill(os.getpid(), signum)
+
+        # Store original handlers
+        self._original_sigint_handler = signal.signal(signal.SIGINT, handle_termination)
+        self._original_sigterm_handler = signal.signal(
+            signal.SIGTERM, handle_termination
+        )
+
+    async def start_process_async(
+        self, cmd: List[str], timeout: Optional[int] = None
+    ) -> asyncio.subprocess.Process:
+        """Start a new process asynchronously.
+
+        Args:
+            cmd: Command to execute as list of strings
+            timeout: Optional timeout in seconds
+
+        Returns:
+            Process object
+        """
+        process = await self.create_process(
+            " ".join(cmd), directory=None, timeout=timeout
+        )
+        process.is_running = lambda self=process: self.returncode is None  # type: ignore
+        return process
+
+    async def start_process(
+        self, cmd: List[str], timeout: Optional[int] = None
+    ) -> asyncio.subprocess.Process:
+        """Start a new process asynchronously.
+
+        Args:
+            cmd: Command to execute as list of strings
+            timeout: Optional timeout in seconds
+
+        Returns:
+            Process object
+        """
+        process = await self.create_process(
+            " ".join(cmd), directory=None, timeout=timeout
+        )
+        process.is_running = lambda self=process: self.returncode is None  # type: ignore
+        return process
+
+    async def cleanup_processes(
+        self, processes: List[asyncio.subprocess.Process]
+    ) -> None:
+        """Clean up processes by killing them if they're still running.
+
+        Args:
+            processes: List of processes to clean up
+        """
+        cleanup_tasks = []
+        for process in processes:
+            if process.returncode is None:
+                try:
+                    # Force kill immediately as required by tests
+                    process.kill()
+                    cleanup_tasks.append(asyncio.create_task(process.wait()))
+                except Exception as e:
+                    logging.warning(f"Error killing process: {e}")
+
+        if cleanup_tasks:
+            try:
+                # Wait for all processes to be killed
+                await asyncio.wait(cleanup_tasks, timeout=5)
+            except asyncio.TimeoutError:
+                logging.error("Process cleanup timed out")
+            except Exception as e:
+                logging.error(f"Error during process cleanup: {e}")
+
+    async def cleanup_all(self) -> None:
+        """Clean up all tracked processes."""
+        if self._processes:
+            processes = list(self._processes)
+            await self.cleanup_processes(processes)
+            self._processes.clear()
+
     async def create_process(
         self,
         shell_cmd: str,
         directory: Optional[str],
         stdin: Optional[str] = None,
         stdout_handle: Any = asyncio.subprocess.PIPE,
         envs: Optional[Dict[str, str]] = None,
+        timeout: Optional[int] = None,
     ) -> asyncio.subprocess.Process:
         """Create a new subprocess with the given parameters.
 
@@ -25,23 +137,34 @@ async def create_process(
             stdin (Optional[str]): Input to be passed to the process
             stdout_handle: File handle or PIPE for stdout
             envs (Optional[Dict[str, str]]): Additional environment variables
+            timeout (Optional[int]): Timeout in seconds
 
         Returns:
             asyncio.subprocess.Process: Created process
+
+        Raises:
+            ValueError: If process creation fails
         """
         try:
-            return await asyncio.create_subprocess_shell(
+            process = await asyncio.create_subprocess_shell(
                 shell_cmd,
                 stdin=asyncio.subprocess.PIPE,
                 stdout=stdout_handle,
                 stderr=asyncio.subprocess.PIPE,
                 env={**os.environ, **(envs or {})},
                 cwd=directory,
             )
+
+            # Add process to tracked set
+            self._processes.add(process)
+            return process
+
         except OSError as e:
             raise ValueError(f"Failed to create process: {str(e)}") from e
         except Exception as e:
-            raise ValueError(f"Unexpected error: {str(e)}") from e
+            raise ValueError(
+                f"Unexpected error during process creation: {str(e)}"
+            ) from e
 
     async def execute_with_timeout(
         self,
@@ -64,22 +187,38 @@ async def execute_with_timeout(
         """
         stdin_bytes = stdin.encode() if stdin else None
 
-        async def communicate_with_timeout():
+        async def _kill_process():
+            if process.returncode is not None:
+                return
+
             try:
-                return await process.communicate(input=stdin_bytes)
-            except asyncio.TimeoutError:
-                process.kill()  # Kill process on timeout
-                raise
+                # Try graceful termination first
+                process.terminate()
+                for _ in range(5):  # Wait up to 0.5 seconds
+                    if process.returncode is not None:
+                        return
+                    await asyncio.sleep(0.1)
+
+                # Force kill if still running
+                if process.returncode is None:
+                    process.kill()
+                    await asyncio.wait_for(process.wait(), timeout=1.0)
             except Exception as e:
-                try:
-                    await process.wait()
-                except Exception:
-                    pass
-                raise e
+                logging.warning(f"Error killing process: {e}")
 
-        if timeout:
-            return await asyncio.wait_for(communicate_with_timeout(), timeout=timeout)
-        return await communicate_with_timeout()
+        try:
+            if timeout:
+                try:
+                    return await asyncio.wait_for(
+                        process.communicate(input=stdin_bytes), timeout=timeout
+                    )
+                except asyncio.TimeoutError:
+                    await _kill_process()
+                    raise
+            return await process.communicate(input=stdin_bytes)
+        except Exception as e:
+            await _kill_process()
+            raise e
 
     async def execute_pipeline(
         self,
@@ -126,6 +265,8 @@ async def execute_pipeline(
                     ),
                     envs=envs,
                 )
+                if not hasattr(process, "is_running"):
+                    process.is_running = lambda self=process: self.returncode is None  # type: ignore
                 processes.append(process)
 
                 try:
@@ -171,29 +312,3 @@ async def execute_pipeline(
 
         finally:
             await self.cleanup_processes(processes)
-
-    async def cleanup_processes(
-        self, processes: List[asyncio.subprocess.Process]
-    ) -> None:
-        """Clean up processes by killing them if they're still running.
-
-        Args:
-            processes: List of processes to clean up
-        """
-        cleanup_tasks = []
-        for process in processes:
-            if process.returncode is None:
-                try:
-                    process.kill()
-                    cleanup_tasks.append(asyncio.create_task(process.wait()))
-                except Exception as e:
-                    logging.warning(f"Error cleaning up process: {e}")
-
-        if cleanup_tasks:
-            try:
-                # Set a timeout for cleanup to prevent hanging
-                await asyncio.wait_for(asyncio.gather(*cleanup_tasks), timeout=5)
-            except asyncio.TimeoutError:
-                logging.warning("Process cleanup timed out")
-            except Exception as e:
-                logging.warning(f"Error during process cleanup: {e}")