Merge pull request #1466 from oraios/lsp-transport

Refactor LanguageServerProcess, separating the communication interface from the concrete implementation

src/solidlsp/language_servers/hlsl_language_server.py

@@ -9,7 +9,6 @@ import pathlib
 import shutil
 from typing import Any, cast
 
-import psutil
 from overrides import override
 
 from solidlsp.ls import (
@@ -253,38 +252,6 @@ class HlslLanguageServer(SolidLanguageServer):
 
         self.server.notify.initialized({})
 
-    @override
-    def stop(self, shutdown_timeout: float = 2.0) -> None:
-        """Kill the shader-language-server process tree before the standard shutdown.
-
-        The base _shutdown() calls process.terminate() directly on the subprocess,
-        which on Windows with shell=True only kills the cmd.exe wrapper, leaving
-        the actual shader-language-server binary running as an orphan. We use psutil
-        to terminate the full process tree first.
-        """
-        process = self.server.process if self.server else None
-        if process and process.pid and process.returncode is None:
-            try:
-                parent = psutil.Process(process.pid)
-                children = parent.children(recursive=True)
-                for child in children:
-                    try:
-                        child.terminate()
-                    except (psutil.NoSuchProcess, psutil.AccessDenied):
-                        pass
-                psutil.wait_procs(children, timeout=2)
-                for child in children:
-                    try:
-                        if child.is_running():
-                            child.kill()
-                    except (psutil.NoSuchProcess, psutil.AccessDenied):
-                        pass
-            except (psutil.NoSuchProcess, psutil.AccessDenied):
-                pass
-            except Exception as e:
-                log.debug(f"Error cleaning up shader-language-server process tree: {e}")
-        super().stop(shutdown_timeout)
-
     @override
     def is_ignored_dirname(self, dirname: str) -> bool:
         """Ignore Unity-specific directories that contain no user-authored shaders."""

src/solidlsp/ls.py

@@ -5,11 +5,10 @@ import logging
 import os
 import pathlib
 import shutil
-import subprocess
 import threading
 from abc import ABC, abstractmethod
 from collections import defaultdict
-from collections.abc import Hashable, Iterator
+from collections.abc import Callable, Hashable, Iterator
 from contextlib import contextmanager
 from copy import copy
 from pathlib import Path, PurePath
@@ -25,7 +24,7 @@ from serena.util.text_utils import MatchedConsecutiveLines
 from solidlsp import ls_types
 from solidlsp.ls_config import FilenameMatcher, Language, LanguageServerConfig
 from solidlsp.ls_exceptions import SolidLSPException
-from solidlsp.ls_process import LanguageServerProcess
+from solidlsp.ls_process import LanguageServerInterface, StdioLanguageServer
 from solidlsp.ls_types import UnifiedSymbolInformation
 from solidlsp.ls_utils import FileUtils, PathUtils, TextUtils
 from solidlsp.lsp_protocol_handler import lsp_types
@@ -336,8 +335,7 @@ class LanguageServerDependencyProviderSinglePath(LanguageServerDependencyProvide
 
 class SolidLanguageServer(ABC):
     """
-    The LanguageServer class provides a language agnostic interface to the Language Server Protocol.
-    It is used to communicate with Language Servers of different programming languages.
+    High-level abstraction for language server interaction, which wraps the underlying low-level LSP interface
     """
 
     CACHE_FOLDER_NAME = "cache"
@@ -500,6 +498,7 @@ class SolidLanguageServer(ABC):
             whenever the format of the raw document symbols changes (typically because the language server
             improves/fixes its output).
         """
+        self._solidlsp_config = config
         self._solidlsp_settings = solidlsp_settings
         lang = self.get_language_enum_instance()
         self._custom_settings = solidlsp_settings.get_ls_specific_settings(lang)
@@ -551,13 +550,10 @@ class SolidLanguageServer(ABC):
             self._dependency_provider = self._create_dependency_provider()
             process_launch_info = self._create_process_launch_info()
         log.debug(f"Creating language server instance with {language_id=} and process launch info: {process_launch_info}")
-        self.server = LanguageServerProcess(
-            process_launch_info,
-            language=self.language,
-            determine_log_level=self._determine_log_level,
-            logger=logging_fn,
-            start_independent_lsp_process=config.start_independent_lsp_process,
-        )
+        self.server = self._create_language_server_interface(process_launch_info, logging_fn)
+        """
+        the low-level language server interface
+        """
         self.server.on_any_notification(self._observe_server_notification)
 
         # Set up the pathspec matcher for the ignored paths
@@ -572,8 +568,6 @@ class SolidLanguageServer(ABC):
         # Create a pathspec matcher from the processed patterns
         self._ignore_spec = pathspec.PathSpec.from_lines(pathspec.patterns.GitWildMatchPattern, processed_patterns)
 
-        self._request_timeout: float | None = None
-
         self._has_waited_for_cross_file_references = False
 
         # resolving additional workspace folders
@@ -599,6 +593,48 @@ class SolidLanguageServer(ABC):
             _seen.add(additional_workspace_abs_path)
             self._additional_workspace_abs_paths.append(additional_workspace_abs_path)
 
+    def _create_dependency_provider(self) -> LanguageServerDependencyProvider:
+        """
+        Creates the dependency provider for this language server.
+
+        Subclasses should override this method to provide their specific dependency provider.
+        This method is only called if process_launch_info is not passed to __init__.
+        """
+        raise NotImplementedError(
+            f"{self.__class__.__name__} must implement _create_dependency_provider() or pass process_launch_info to __init__()"
+        )
+
+    def _create_process_launch_info(self) -> ProcessLaunchInfo:
+        assert self._dependency_provider is not None
+        cmd = self._dependency_provider.create_launch_command()
+        env = self._dependency_provider.create_launch_command_env()
+        return ProcessLaunchInfo(cmd=cmd, cwd=self.repository_root_path, env=env)
+
+    def _create_language_server_interface(
+        self, process_launch_info: ProcessLaunchInfo, logging_fn: Callable[[str, str, StringDict | str], None] | None
+    ) -> LanguageServerInterface:
+        """
+        Creates the low-level language server interface for LSP communication.
+
+        The interface is created but not started.
+
+        The default implementation creates a StdioLanguageServer, but subclasses can override this method to create a different type
+        of interface if needed.
+
+        :param process_launch_info: process launch information
+        :param logging_fn: the trace logging function
+        :return: the interface
+        """
+        return StdioLanguageServer(
+            process_launch_info,
+            language=self.language,
+            determine_log_level=self._determine_log_level,
+            logger=logging_fn,
+            start_independent_lsp_process=self._solidlsp_config.start_independent_lsp_process,
+        )
+
+    # --- diagnostics-related functions ---
+
     def _observe_server_notification(self, method: str, params: Any) -> None:
         """
         Observe notifications sent by the language server.
@@ -981,23 +1017,6 @@ class SolidLanguageServer(ABC):
 
         return self._filter_diagnostics(ret, start_line, end_line, min_severity)
 
-    def _create_dependency_provider(self) -> LanguageServerDependencyProvider:
-        """
-        Creates the dependency provider for this language server.
-
-        Subclasses should override this method to provide their specific dependency provider.
-        This method is only called if process_launch_info is not passed to __init__.
-        """
-        raise NotImplementedError(
-            f"{self.__class__.__name__} must implement _create_dependency_provider() or pass process_launch_info to __init__()"
-        )
-
-    def _create_process_launch_info(self) -> ProcessLaunchInfo:
-        assert self._dependency_provider is not None
-        cmd = self._dependency_provider.create_launch_command()
-        env = self._dependency_provider.create_launch_command_env()
-        return ProcessLaunchInfo(cmd=cmd, cwd=self.repository_root_path, env=env)
-
     def _get_wait_time_for_cross_file_referencing(self) -> float:
         """Meant to be overridden by subclasses for LS that don't have a reliable "finished initializing" signal.
 
@@ -1187,78 +1206,20 @@ class SolidLanguageServer(ABC):
 
         return match_path(relative_path, self.get_ignore_spec(), root_path=self.repository_root_path)
 
-    def _shutdown(self, timeout: float = 5.0) -> None:
-        """
-        A robust shutdown process designed to terminate cleanly on all platforms, including Windows,
-        by explicitly closing all I/O pipes.
-        """
-        if not self.server.is_running():
-            log.debug("Server process not running, skipping shutdown.")
-            return
-
-        log.info(f"Initiating final robust shutdown with a {timeout}s timeout...")
-        process = self.server.process
-        if process is None:
-            log.debug("Server process is None, cannot shutdown.")
-            return
-
-        # --- Main Shutdown Logic ---
-        # Stage 1: Graceful Termination Request
-        # Send LSP shutdown and close stdin to signal no more input.
-        try:
-            log.debug("Sending LSP shutdown request...")
-            # Use a thread to timeout the LSP shutdown call since it can hang
-            shutdown_thread = threading.Thread(target=self.server.shutdown)
-            shutdown_thread.daemon = True
-            shutdown_thread.start()
-            shutdown_thread.join(timeout=2.0)  # 2 second timeout for LSP shutdown
-
-            if shutdown_thread.is_alive():
-                log.debug("LSP shutdown request timed out, proceeding to terminate...")
-            else:
-                log.debug("LSP shutdown request completed.")
-
-            if process.stdin and not process.stdin.closed:
-                process.stdin.close()
-            log.debug("Stage 1 shutdown complete.")
-        except Exception as e:
-            log.debug(f"Exception during graceful shutdown: {e}")
-            # Ignore errors here, we are proceeding to terminate anyway.
-
-        # Stage 2: Terminate and Wait for Process to Exit
-        log.debug(f"Terminating process {process.pid}, current status: {process.poll()}")
-        process.terminate()
-
-        # Stage 3: Wait for process termination with timeout
-        try:
-            log.debug(f"Waiting for process {process.pid} to terminate...")
-            exit_code = process.wait(timeout=timeout)
-            log.info(f"Language server process terminated successfully with exit code {exit_code}.")
-        except subprocess.TimeoutExpired:
-            # If termination failed, forcefully kill the process
-            log.warning(f"Process {process.pid} termination timed out, killing process forcefully...")
-            process.kill()
-            try:
-                exit_code = process.wait(timeout=2.0)
-                log.info(f"Language server process killed successfully with exit code {exit_code}.")
-            except subprocess.TimeoutExpired:
-                log.error(f"Process {process.pid} could not be killed within timeout.")
-        except Exception as e:
-            log.error(f"Error during process shutdown: {e}")
-
     @contextmanager
-    def start_server(self) -> Iterator["SolidLanguageServer"]:
+    def start_server_context(self) -> Iterator["SolidLanguageServer"]:
         self.start()
         yield self
         self.stop()
 
-    def _start_server_process(self) -> None:
-        self.server_started = True
-        self._start_server()
-
     @abstractmethod
     def _start_server(self) -> None:
-        pass
+        """
+        Starts the low-level language server interface (i.e. self.server).
+
+        This method must ultimately call `self.server.start()` and may perform additional setup before or after starting the server,
+        such as waiting for initialization to complete or activating additional workspaces.
+        """
 
     def _get_language_id_for_file(self, relative_file_path: str) -> str:
         """
@@ -3111,7 +3072,8 @@ class SolidLanguageServer(ABC):
         :return: self for method chaining
         """
         log.info(f"Starting language server with language {self.language_server.language} for {self.language_server.repository_root_path}")
-        self._start_server_process()
+        self.server_started = True
+        self._start_server()
         return self
 
     def stop(self, shutdown_timeout: float = 2.0) -> None:
@@ -3122,16 +3084,18 @@ class SolidLanguageServer(ABC):
         :param shutdown_timeout: time, in seconds, to wait for the server to shutdown gracefully before killing it
         """
         try:
-            self._shutdown(timeout=shutdown_timeout)
+            self.server.stop(timeout=shutdown_timeout)
         except Exception as e:
             log.warning(f"Exception while shutting down language server: {e}")
+        finally:
+            self.server_started = False
 
     @property
     def language_server(self) -> Self:
         return self
 
     @property
-    def handler(self) -> LanguageServerProcess:
+    def handler(self) -> LanguageServerInterface:
         """Access the underlying language server handler.
 
         Useful for advanced operations like sending custom commands

src/solidlsp/ls_process.py

@@ -6,12 +6,12 @@ import platform
 import subprocess
 import threading
 import time
+from abc import ABC, abstractmethod
 from collections.abc import Callable
 from dataclasses import dataclass
 from queue import Empty, Queue
-from typing import Any
+from typing import IO, Any, AnyStr
 
-import psutil
 from sensai.util.string import ToStringMixin
 
 from solidlsp.ls_config import Language
@@ -32,7 +32,7 @@ from solidlsp.lsp_protocol_handler.server import (
     make_request,
     make_response,
 )
-from solidlsp.util.subprocess_util import quote_arg, subprocess_kwargs
+from solidlsp.util.subprocess_util import quote_arg, subprocess_kwargs, terminate_process_tree_with_kill_fallback
 
 log = logging.getLogger(__name__)
 
@@ -92,74 +92,65 @@ class Request(ToStringMixin):
             raise e
 
 
-class LanguageServerProcess:
+class LanguageServerInterface(ABC):
     """
-    Represents a language server process and provides methods for communicating with it using the
+    Represents an interface to a language server, providing methods for communicating with it using the
     Language Server Protocol (LSP).
 
     It provides methods for sending requests, responses, and notifications to the server
     and for registering handlers for requests and notifications from the server.
 
     Uses JSON-RPC 2.0 for communication with the server over stdin/stdout.
-
-    Attributes:
-        send: A LspRequest object that can be used to send requests to the server and
-            await for the responses.
-        notify: A LspNotification object that can be used to send notifications to the server.
-        cmd: A string that represents the command to launch the language server process.
-        process: A subprocess.Popen object that represents the language server process.
-        request_id: An integer that represents the next available request id for the client.
-        _pending_requests: A dictionary that maps request ids to Request objects that
-            store the results or errors of the requests.
-        on_request_handlers: A dictionary that maps method names to callback functions
-            that handle requests from the server.
-        on_notification_handlers: A dictionary that maps method names to callback functions
-            that handle notifications from the server.
-        _trace_log_fn: An optional function that takes two strings (source and destination) and
-            a payload dictionary, and logs the communication between the client and the server.
-        tasks: A dictionary that maps task ids to asyncio.Task objects that represent
-            the asynchronous tasks created by the handler.
-        task_counter: An integer that represents the next available task id for the handler.
-        loop: An asyncio.AbstractEventLoop object that represents the event loop used by the handler.
-        start_independent_lsp_process: An optional boolean flag that indicates whether to start the
-        language server process in an independent process group. Default is `True`. Setting it to
-        `False` means that the language server process will be in the same process group as the
-        the current process, and any SIGINT and SIGTERM signals will be sent to both processes.
-
     """
 
     def __init__(
         self,
-        process_launch_info: ProcessLaunchInfo,
         language: Language,
         determine_log_level: Callable[[str], int],
         logger: Callable[[str, str, StringDict | str], None] | None = None,
-        start_independent_lsp_process: bool = True,
         request_timeout: float | None = None,
     ) -> None:
+        """
+        :param language: the language
+        :param determine_log_level: a function for log lines read from stderr, which determines the log level
+        :param logger: the trace logger function
+        :param request_timeout: the timeout, in seconds, for all requests sent to the language server. If None, no timeout will be applied.
+        """
         self.language = language
         self._determine_log_level = determine_log_level
         self.send = LanguageServerRequest(self)
+        """
+        an object that can be used to send requests to the server 
+        """
         self.notify = LspNotification(self.send_notification)
-
-        self.process_launch_info = process_launch_info
-        self.process: subprocess.Popen[bytes] | None = None
-        self._is_shutting_down = False
-
+        """
+        an object that can be used to send notifications to the server
+        """
         self.request_id = 1
+        """
+        the next request id to use for requests
+        """
         self._pending_requests: dict[Any, Request] = {}
+        """
+        maps request ids to Request objects that store the results or errors of the requests
+        """
         self.on_request_handlers: dict[str, Callable[[Any], Any]] = {}
         self.on_notification_handlers: dict[str, Callable[[Any], None]] = {}
+        """
+        maps method names to callback functions that handle notifications from the server
+        """
         self._notification_observers: list[Callable[[str, Any], None]] = []
         self._trace_log_fn = logger
-        self.tasks: dict[int, Any] = {}
         self.task_counter = 0
-        self.loop = None
-        self.start_independent_lsp_process = start_independent_lsp_process
+        self._is_stopping = False
+        """
+        Flag indicating whether the interface is in the process of stopping (or already stopped).
+        Exception handlers should check this flag to avoid logging errors that are caused by the interface being stopped.
+        """
+        self._incoming_messages_queue: Queue[bytes] = Queue()
         self._request_timeout = request_timeout
 
         # Add thread locks for shared resources to prevent race conditions
-        self._stdin_lock = threading.Lock()
         self._request_id_lock = threading.Lock()
         self._response_handlers_lock = threading.Lock()
         self._tasks_lock = threading.Lock()
@@ -170,136 +161,50 @@ class LanguageServerProcess:
         """
         self._request_timeout = timeout
 
+    @abstractmethod
     def is_running(self) -> bool:
         """
-        Checks if the language server process is currently running.
+        Checks whether the language server interface is running
         """
-        return self.process is not None and self.process.returncode is None
 
     def start(self) -> None:
         """
-        Starts the language server process and creates a task to continuously read from its stdout to handle communications
-        from the server to the client
+        Starts communication with the language server
         """
-        child_proc_env = os.environ.copy()
-        child_proc_env.update(self.process_launch_info.env)
+        self._is_stopping = False
+        self._start()
 
-        cmd = self.process_launch_info.cmd
-        is_windows = platform.system() == "Windows"
-        if not isinstance(cmd, str) and not is_windows:
-            # Since we are using the shell, we need to convert the command list to a single string
-            # on Linux/macOS
-            cmd = " ".join(map(quote_arg, cmd))
-        log.info("Starting language server process via command: %s", self.process_launch_info.cmd)
-        kwargs = subprocess_kwargs()
-        kwargs["start_new_session"] = self.start_independent_lsp_process
-        self.process = subprocess.Popen(
-            cmd,
-            stdout=subprocess.PIPE,
-            stdin=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-            env=child_proc_env,
-            cwd=self.process_launch_info.cwd,
-            shell=True,
-            **kwargs,
-        )
-
-        # Check if process terminated immediately
-        if self.process.returncode is not None:
-            log.error("Language server has already terminated/could not be started")
-            # Process has already terminated
-            stderr_data = self.process.stderr.read() if self.process.stderr else b""
-            error_message = stderr_data.decode("utf-8", errors="replace")
-            raise RuntimeError(f"Process terminated immediately with code {self.process.returncode}. Error: {error_message}")
-
-        # start threads to read stdout and stderr of the process
-        threading.Thread(
-            target=self._read_ls_process_stdout,
-            name=f"LSP-stdout-reader:{self.language.value}",
-            daemon=True,
-        ).start()
-        threading.Thread(
-            target=self._read_ls_process_stderr,
-            name=f"LSP-stderr-reader:{self.language.value}",
-            daemon=True,
-        ).start()
-
-    def stop(self) -> None:
+    @abstractmethod
+    def _start(self) -> None:
         """
-        Sends the terminate signal to the language server process and waits for it to exit, with a timeout, killing it if necessary
+        Starts the actual communication mechanism with the language server, calling `_handle_body` for each message received from the
+        language server
         """
-        process = self.process
-        self.process = None
-        if process:
-            self._cleanup_process(process)
 
-    def _cleanup_process(self, process: subprocess.Popen[bytes]) -> None:
-        """Clean up a process: close stdin, terminate/kill process, close stdout/stderr."""
-        # Close stdin first to prevent deadlocks
-        # See: https://bugs.python.org/issue35539
-        self._safely_close_pipe(process.stdin)
-
-        # Terminate/kill the process if it's still running
-        if process.returncode is None:
-            self._terminate_or_kill_process(process)
-
-        # Close stdout and stderr pipes after process has exited
-        # This is essential to prevent "I/O operation on closed pipe" errors and
-        # "Event loop is closed" errors during garbage collection
-        # See: https://bugs.python.org/issue41320 and https://github.com/python/cpython/issues/88050
-        self._safely_close_pipe(process.stdout)
-        self._safely_close_pipe(process.stderr)
-
-    def _safely_close_pipe(self, pipe: Any) -> None:
-        """Safely close a pipe, ignoring any exceptions."""
-        if pipe:
-            try:
-                pipe.close()
-            except Exception:
-                pass
-
-    def _terminate_or_kill_process(self, process: subprocess.Popen[bytes]) -> None:
-        """Try to terminate the process gracefully, then forcefully if necessary."""
-        # First try to terminate the process tree gracefully
-        self._signal_process_tree(process, terminate=True)
-
-    def _signal_process_tree(self, process: subprocess.Popen[bytes], terminate: bool = True) -> None:
-        """Send signal (terminate or kill) to the process and all its children."""
-        signal_method = "terminate" if terminate else "kill"
-
-        # Try to get the parent process
-        parent = None
-        try:
-            parent = psutil.Process(process.pid)
-        except (psutil.NoSuchProcess, psutil.AccessDenied, Exception):
-            pass
-
-        # If we have the parent process and it's running, signal the entire tree
-        if parent and parent.is_running():
-            # Signal children first
-            for child in parent.children(recursive=True):
-                try:
-                    getattr(child, signal_method)()
-                except (psutil.NoSuchProcess, psutil.AccessDenied, Exception):
-                    pass
-
-            # Then signal the parent
-            try:
-                getattr(parent, signal_method)()
-            except (psutil.NoSuchProcess, psutil.AccessDenied, Exception):
-                pass
-        else:
-            # Fall back to direct process signaling
-            try:
-                getattr(process, signal_method)()
-            except Exception:
-                pass
-
-    def shutdown(self) -> None:
+    def stop(self, timeout: float = 5.0) -> None:
         """
-        Perform the shutdown sequence for the client, including sending the shutdown request to the server and notifying it of exit
+        Terminates communication with the language server, freeing resources
+
+        :param timeout: the maximum time, in seconds, to wait for the language server to terminate before forcefully killing it
+            (if applicable)
+        """
+        if not self.is_running():
+            log.debug("Server process not running, skipping shutdown.")
+            return
+
+        self._is_stopping = True
+
+        log.info(f"Initiating shutdown with a {timeout}s timeout...")
+        self._stop(timeout)
+
+    @abstractmethod
+    def _stop(self, timeout: float) -> None:
+        pass
+
+    def _send_shutdown(self) -> None:
+        """
+        Signals shutdown to the server
         """
-        self._is_shutting_down = True
         log.info("Sending shutdown request to server")
         self.send.shutdown()
         log.info("Received shutdown response from server")
@@ -307,6 +212,23 @@ class LanguageServerProcess:
         self.notify.exit()
         log.info("Sent exit notification to server")
 
+    def _send_shutdown_in_thread(self) -> None:
+        """
+        Signals shutdown to the server in a separate thread (requests can hang),
+        and waits for the thread to complete with a timeout.
+
+        :param timeout: timeout, in seconds, to wait for the requests to be handled
+        """
+        log.debug("Sending LSP shutdown request...")
+        shutdown_thread = threading.Thread(target=self._send_shutdown)
+        shutdown_thread.daemon = True
+        shutdown_thread.start()
+        shutdown_thread.join(timeout=2.0)
+        if shutdown_thread.is_alive():
+            log.debug("LSP shutdown request timed out, proceeding to terminate...")
+        else:
+            log.debug("LSP shutdown request completed.")
+
     def _trace(self, src: str, dest: str, message: str | StringDict) -> None:
         """
         Traces LS communication by logging the message with the source and destination of the message
@@ -314,89 +236,9 @@ class LanguageServerProcess:
         if self._trace_log_fn is not None:
             self._trace_log_fn(src, dest, message)
 
-    def _read_bytes_from_process(self, process, stream, num_bytes) -> bytes:  # type: ignore
-        """Read exactly num_bytes from process stdout"""
-        data = b""
-        while len(data) < num_bytes:
-            chunk = stream.read(num_bytes - len(data))
-            if not chunk:
-                if process.poll() is not None:
-                    raise LanguageServerTerminatedException(
-                        f"Process terminated while trying to read response (read {len(data)} of {num_bytes} bytes before termination)",
-                        language=self.language,
-                    )
-                # Process still running but no data available yet, retry after a short delay
-                time.sleep(0.01)
-                continue
-            data += chunk
-        return data
-
-    def _read_ls_process_stdout(self) -> None:
-        """
-        Continuously read from the language server process stdout and handle the messages
-        invoking the registered response and notification handlers
-        """
-        exception: Exception | None = None
-        try:
-            while self.process and self.process.stdout:
-                if self.process.poll() is not None:  # process has terminated
-                    break
-                line = self.process.stdout.readline()
-                if not line:
-                    continue
-                try:
-                    num_bytes = content_length(line)
-                except ValueError:
-                    continue
-                if num_bytes is None:
-                    continue
-                while line and line.strip():
-                    line = self.process.stdout.readline()
-                if not line:
-                    continue
-                body = self._read_bytes_from_process(self.process, self.process.stdout, num_bytes)
-
-                self._handle_body(body)
-        except LanguageServerTerminatedException as e:
-            exception = e
-        except (BrokenPipeError, ConnectionResetError) as e:
-            exception = LanguageServerTerminatedException("Language server process terminated while reading stdout", self.language, cause=e)
-        except Exception as e:
-            exception = LanguageServerTerminatedException(
-                "Unexpected error while reading stdout from language server process", self.language, cause=e
-            )
-        log.info("Language server stdout reader thread has terminated")
-        if not self._is_shutting_down:
-            if exception is None:
-                exception = LanguageServerTerminatedException("Language server stdout read process terminated unexpectedly", self.language)
-            log.error(str(exception))
-            self._cancel_pending_requests(exception)
-
-    def _read_ls_process_stderr(self) -> None:
-        """
-        Continuously read from the language server process stderr and log the messages
-        """
-        try:
-            while self.process and self.process.stderr:
-                if self.process.poll() is not None:
-                    # process has terminated
-                    break
-                line = self.process.stderr.readline()
-                if not line:
-                    continue
-                line_str = line.decode(ENCODING, errors="replace")
-                level = self._determine_log_level(line_str)
-                log.log(level, line_str)
-        except Exception as e:
-            log.error("Error while reading stderr from language server process: %s", e, exc_info=e)
-        if not self._is_shutting_down:
-            log.error("Language server stderr reader thread terminated unexpectedly")
-        else:
-            log.info("Language server stderr reader thread has terminated")
-
     def _handle_body(self, body: bytes) -> None:
         """
-        Parse the body text received from the language server process and invoke the appropriate handler
+        Parses the body text received from the language server process and invokes the appropriate handler
         """
         try:
             self._receive_payload(json.loads(body))
@@ -480,24 +322,11 @@ class LanguageServerProcess:
         log.debug("Returning result:\n%s", result.payload)
         return result.payload
 
+    @abstractmethod
     def _send_payload(self, payload: StringDict) -> None:
         """
-        Send the payload to the server by writing to its stdin asynchronously.
+        Send the given payload to the server
         """
-        if not self.process or not self.process.stdin:
-            return
-        self._trace("solidlsp", "ls", payload)
-        msg = create_message(payload)
-
-        # Use lock to prevent concurrent writes to stdin that cause buffer corruption
-        with self._stdin_lock:
-            try:
-                self.process.stdin.writelines(msg)
-                self.process.stdin.flush()
-            except (BrokenPipeError, ConnectionResetError, OSError) as e:
-                # Log the error but don't raise to prevent cascading failures
-                log.error(f"Failed to write to stdin: {e}")
-                return
 
     def on_request(self, method: str, cb: Callable[[Any], Any]) -> None:
         """
@@ -575,7 +404,7 @@ class LanguageServerProcess:
             except asyncio.CancelledError:
                 return
             except Exception as ex:
-                if not self._is_shutting_down:
+                if not self._is_stopping:
                     log.error("Error handling notification observer for method '%s': %s", method, ex, exc_info=ex)
 
         handler = self.on_notification_handlers.get(method)
@@ -587,5 +416,207 @@ class LanguageServerProcess:
         except asyncio.CancelledError:
             return
         except Exception as ex:
-            if not self._is_shutting_down:
+            if not self._is_stopping:
                 log.error("Error handling notification for method '%s': %s", method, ex, exc_info=ex)
+
+
+class StdioLanguageServer(LanguageServerInterface):
+    """
+    Represents a language server interface where the language server is launched as a subprocess
+    and communication takes place over the process' stdin/stdout streams.
+    """
+
+    def __init__(
+        self,
+        process_launch_info: ProcessLaunchInfo,
+        language: Language,
+        determine_log_level: Callable[[str], int],
+        logger: Callable[[str, str, StringDict | str], None] | None = None,
+        start_independent_lsp_process: bool = True,
+        request_timeout: float | None = None,
+    ) -> None:
+        """
+        :param process_launch_info: the information required to launch the language server process
+        :param language: the language
+        :param determine_log_level: a function for log lines read from stderr, which determines the log level
+        :param logger: the trace logger function
+        :param start_independent_lsp_process: whether to start the language server process in an independent process group
+        :param request_timeout: the timeout, in seconds, for all requests sent to the language server. If None, no timeout will be applied.
+        """
+        super().__init__(language, determine_log_level, logger, request_timeout)
+
+        self.process_launch_info = process_launch_info
+        self.process: subprocess.Popen[bytes] | None = None
+        self.start_independent_lsp_process = start_independent_lsp_process
+
+        self._stdin_lock = threading.Lock()
+
+    def is_running(self) -> bool:
+        return self.process is not None and self.process.returncode is None
+
+    def _start(self) -> None:
+        child_proc_env = os.environ.copy()
+        child_proc_env.update(self.process_launch_info.env)
+
+        cmd = self.process_launch_info.cmd
+        is_windows = platform.system() == "Windows"
+        if not isinstance(cmd, str) and not is_windows:
+            # Since we are using the shell, we need to convert the command list to a single string
+            # on Linux/macOS
+            cmd = " ".join(map(quote_arg, cmd))
+        log.info("Starting language server process via command: %s", self.process_launch_info.cmd)
+        kwargs = subprocess_kwargs()
+        kwargs["start_new_session"] = self.start_independent_lsp_process
+        self.process = subprocess.Popen(
+            cmd,
+            stdout=subprocess.PIPE,
+            stdin=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            env=child_proc_env,
+            cwd=self.process_launch_info.cwd,
+            shell=True,
+            **kwargs,
+        )
+
+        # Check if process terminated immediately
+        if self.process.returncode is not None:
+            log.error("Language server has already terminated/could not be started")
+            # Process has already terminated
+            stderr_data = self.process.stderr.read() if self.process.stderr else b""
+            error_message = stderr_data.decode("utf-8", errors="replace")
+            raise RuntimeError(f"Process terminated immediately with code {self.process.returncode}. Error: {error_message}")
+
+        # start threads to read stdout and stderr of the process
+        threading.Thread(
+            target=self._read_ls_process_stdout,
+            name=f"LSP-stdout-reader:{self.language.value}",
+            daemon=True,
+        ).start()
+        threading.Thread(
+            target=self._read_ls_process_stderr,
+            name=f"LSP-stderr-reader:{self.language.value}",
+            daemon=True,
+        ).start()
+
+    def _stop(self, timeout: float) -> None:
+        if self.process is None:
+            log.debug("Server process is None, cannot shutdown.")
+            return
+
+        try:
+            # send LSP shutdown and close stdin to signal no more input
+            try:
+                self._send_shutdown_in_thread()
+                self._safely_close_pipe(self.process.stdin)
+            except Exception as e:
+                log.debug(f"Exception during graceful shutdown: {e}")
+                # Ignore errors here, we are proceeding to terminate anyway.
+            # terminate the process
+            terminate_process_tree_with_kill_fallback(self.process, terminate_timeout=timeout, process_name=f"LS[{self.language.value}]")
+        finally:
+            self.process = None
+
+    @staticmethod
+    def _safely_close_pipe(pipe: IO[AnyStr] | None) -> None:
+        """Safely close a pipe, ignoring any exceptions."""
+        if pipe and not pipe.closed:
+            try:
+                pipe.close()
+            except Exception:
+                pass
+
+    def _read_bytes_from_process(self, process, stream, num_bytes) -> bytes:  # type: ignore
+        """Read exactly num_bytes from process stdout"""
+        data = b""
+        while len(data) < num_bytes:
+            chunk = stream.read(num_bytes - len(data))
+            if not chunk:
+                if process.poll() is not None:
+                    raise LanguageServerTerminatedException(
+                        f"Process terminated while trying to read response (read {len(data)} of {num_bytes} bytes before termination)",
+                        language=self.language,
+                    )
+                # Process still running but no data available yet, retry after a short delay
+                time.sleep(0.01)
+                continue
+            data += chunk
+        return data
+
+    def _read_ls_process_stdout(self) -> None:
+        """
+        Continuously read from the language server process stdout and handle the messages
+        invoking the registered response and notification handlers
+        """
+        exception: Exception | None = None
+        try:
+            while self.process and self.process.stdout:
+                if self.process.poll() is not None:  # process has terminated
+                    break
+                line = self.process.stdout.readline()
+                if not line:
+                    continue
+                try:
+                    num_bytes = content_length(line)
+                except ValueError:
+                    continue
+                if num_bytes is None:
+                    continue
+                while line and line.strip():
+                    line = self.process.stdout.readline()
+                if not line:
+                    continue
+                body = self._read_bytes_from_process(self.process, self.process.stdout, num_bytes)
+
+                self._handle_body(body)
+        except LanguageServerTerminatedException as e:
+            exception = e
+        except (BrokenPipeError, ConnectionResetError) as e:
+            exception = LanguageServerTerminatedException("Language server process terminated while reading stdout", self.language, cause=e)
+        except Exception as e:
+            exception = LanguageServerTerminatedException(
+                "Unexpected error while reading stdout from language server process", self.language, cause=e
+            )
+        log.info("Language server stdout reader thread has terminated")
+        if not self._is_stopping:
+            if exception is None:
+                exception = LanguageServerTerminatedException("Language server stdout read process terminated unexpectedly", self.language)
+            log.error(str(exception))
+            self._cancel_pending_requests(exception)
+
+    def _read_ls_process_stderr(self) -> None:
+        """
+        Continuously read from the language server process stderr and log the messages
+        """
+        try:
+            while self.process and self.process.stderr:
+                if self.process.poll() is not None:
+                    # process has terminated
+                    break
+                line = self.process.stderr.readline()
+                if not line:
+                    continue
+                line_str = line.decode(ENCODING, errors="replace")
+                level = self._determine_log_level(line_str)
+                log.log(level, line_str)
+        except Exception as e:
+            log.error("Error while reading stderr from language server process: %s", e, exc_info=e)
+        if not self._is_stopping:
+            log.error("Language server stderr reader thread terminated unexpectedly")
+        else:
+            log.info("Language server stderr reader thread has terminated")
+
+    def _send_payload(self, payload: StringDict) -> None:
+        if not self.process or not self.process.stdin:
+            return
+        self._trace("solidlsp", "ls", payload)
+        msg = create_message(payload)
+
+        # Use lock to prevent concurrent writes to stdin that cause buffer corruption
+        with self._stdin_lock:
+            try:
+                self.process.stdin.writelines(msg)
+                self.process.stdin.flush()
+            except (BrokenPipeError, ConnectionResetError, OSError) as e:
+                # Log the error but don't raise to prevent cascading failures
+                log.error(f"Failed to write to stdin: {e}")
+                return

src/solidlsp/ls_request.py

@@ -3,11 +3,11 @@ from typing import TYPE_CHECKING, Any, Union
 from solidlsp.lsp_protocol_handler import lsp_types
 
 if TYPE_CHECKING:
-    from .ls_process import LanguageServerProcess
+    from .ls_process import LanguageServerInterface
 
 
 class LanguageServerRequest:
-    def __init__(self, handler: "LanguageServerProcess"):
+    def __init__(self, handler: "LanguageServerInterface"):
         self.handler = handler
 
     def _send_request(self, method: str, params: Any | None = None) -> Any:

src/solidlsp/util/subprocess_util.py

@@ -1,7 +1,12 @@
+import logging
 import platform
 import shlex
 import subprocess
 
+import psutil
+
+log = logging.getLogger(__name__)
+
 
 def subprocess_kwargs() -> dict:
     """
@@ -27,3 +32,66 @@ def quote_arg(arg: str) -> str:
             return arg
         return f'"{arg}"'
     return shlex.quote(arg)
+
+
+def _signal_process_tree(process: subprocess.Popen[bytes], terminate: bool = True) -> None:
+    """
+    Sends a signal (terminate or kill) to the given process and all its children.
+
+    :param terminate: if True, signal terminate, otherwise signal kill
+    """
+
+    def signal_process(p: subprocess.Popen | psutil.Process) -> None:
+        try:
+            if terminate:
+                p.terminate()
+            else:
+                p.kill()
+        except:
+            pass
+
+    # Try to get the parent process
+    parent = None
+    try:
+        parent = psutil.Process(process.pid)
+    except (psutil.NoSuchProcess, psutil.AccessDenied, Exception):
+        pass
+
+    # If we have the parent process and it's running, signal the entire tree
+    if parent and parent.is_running():
+        for child in parent.children(recursive=True):
+            signal_process(child)
+        signal_process(parent)
+    # Otherwise, fall back to direct process signaling
+    else:
+        signal_process(process)
+
+
+def terminate_process_tree_with_kill_fallback(process: subprocess.Popen, terminate_timeout: float, process_name: str = "Process") -> None:
+    """
+    Attempts to terminate the given process and its children by signaling them to terminate,
+    and if that fails (i.e. they don't exit within the given timeout), forcefully kills them.
+
+    The termination is logged.
+
+    :param process: the process to terminate
+    :param terminate_timeout: the time to wait for the process to terminate gracefully before killing it
+    :param process_name: the name of the process (used for logging purposes); should start with capital letter
+    """
+    log.debug(f"Terminating process {process.pid}, current status: {process.poll()}")
+    _signal_process_tree(process, terminate=True)
+    try:
+        log.debug(f"Waiting for process {process.pid} to terminate...")
+        exit_code = process.wait(timeout=terminate_timeout)
+        log.info(f"{process_name} terminated successfully with exit code {exit_code}.")
+    except subprocess.TimeoutExpired:
+        # If termination failed, forcefully kill the process
+        log.warning(f"{process_name} (pid={process.pid}) termination timed out, killing process forcefully...")
+        _signal_process_tree(process, terminate=False)
+        try:
+            exit_code = process.wait(timeout=2.0)
+            log.info(f"{process_name} killed successfully with exit code {exit_code}.")
+        except subprocess.TimeoutExpired:
+            log.error(f"{process_name} (pid={process.pid}) could not be killed within timeout.")
+    except Exception as e:
+        log.error(f"Error during process shutdown: {e}")

test/conftest.py

@@ -103,22 +103,8 @@ def start_ls_context(
         language, repo_path, ignored_paths, trace_lsp_communication, ls_specific_settings, additional_workspace_folders, solidlsp_dir
     )
     log.info(f"Starting language server for {language} {repo_path}")
-    ls.start()
-    try:
-        log.info(f"Language server started for {language} {repo_path}")
+    with ls.start_server_context():
         yield ls
-    finally:
-        log.info(f"Stopping language server for {language} {repo_path}")
-        try:
-            ls.stop(shutdown_timeout=5)
-        except Exception as e:
-            log.warning(f"Warning: Error stopping language server: {e}")
-            # try to force cleanup
-            if hasattr(ls, "server") and hasattr(ls.server, "process"):
-                try:
-                    ls.server.process.terminate()
-                except:
-                    pass
 
 
 @contextmanager

test/solidlsp/scala/test_scala_language_server.py

@@ -22,7 +22,7 @@ def scala_ls():
     solidlsp_settings = SolidLSPSettings()
     ls = ScalaLanguageServer(config, repo_root, solidlsp_settings)
 
-    with ls.start_server():
+    with ls.start_server_context():
         yield ls