diff --git a/src/solidlsp/dependency_provider.py b/src/solidlsp/dependency_provider.py new file mode 100644 index 00000000..f296fee6 --- /dev/null +++ b/src/solidlsp/dependency_provider.py @@ -0,0 +1,3367 @@ +import dataclasses +import hashlib +import json +import logging +import os +import pathlib +import shutil +import threading +from abc import ABC, abstractmethod +from collections import defaultdict +from collections.abc import Callable, Hashable, Iterator, Sequence +from contextlib import contextmanager +from copy import copy +from dataclasses import dataclass +from pathlib import Path, PurePath +from time import monotonic, perf_counter, sleep +from typing import Any, Self, Union, cast + +import pathspec +from sensai.util.pickle import getstate, load_pickle +from sensai.util.string import ToStringMixin + +from serena.util.file_system import match_path +from serena.util.text_utils import MatchedConsecutiveLines +from solidlsp import ls_types +from solidlsp.initialize_params import DefaultInitializeParamsBuilder, InitializeParamsBuilder +from solidlsp.ls_config import FilenameMatcher, Language, LanguageServerConfig +from solidlsp.ls_exceptions import SolidLSPException +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 +from solidlsp.lsp_protocol_handler import lsp_types as LSPTypes +from solidlsp.lsp_protocol_handler.lsp_constants import LSPConstants +from solidlsp.lsp_protocol_handler.lsp_types import ( + Definition, + DefinitionParams, + DocumentSymbol, + ImplementationParams, + InitializeParams, + LocationLink, + RenameParams, + SymbolInformation, +) +from solidlsp.lsp_protocol_handler.server import ( + LSPError, + ProcessLaunchInfo, + StringDict, +) +from solidlsp.settings import SolidLSPSettings +from solidlsp.util.cache import load_cache, save_cache + +RawDocumentSymbol = Union[DocumentSymbol, SymbolInformation] +""" +Type alias for the raw symbol information returned by a language server in response to a +`textDocument/documentSymbol` request. +The `DocumentSymbol` is the preferred type, but the legacy type `SymbolInformation` is also still used. +""" + +log = logging.getLogger(__name__) + +_debug_enabled = log.isEnabledFor(logging.DEBUG) +"""Serves as a flag that triggers additional computation when debug logging is enabled.""" + + +@dataclasses.dataclass(kw_only=True) +class ReferenceInSymbol: + """A symbol retrieved when requesting reference to a symbol, together with the location of the reference""" + + symbol: ls_types.UnifiedSymbolInformation + line: int + character: int + + +class LSPFileBuffer: + """ + This class is used to store the contents of an open LSP file in memory. + """ + + def __init__( + self, + abs_path: Path, + uri: str, + encoding: str, + version: int, + language_id: str, + ref_count: int, + language_server: "SolidLanguageServer", + open_in_ls: bool = True, + ) -> None: + self.abs_path = abs_path + self.language_server = language_server + self.uri = uri + self._read_file_modified_date: float | None = None + self._contents: str | None = None + self.version = version + self.language_id = language_id + self.ref_count = ref_count + self.encoding = encoding + self._content_hash: str | None = None + self._is_open_in_ls = False + if open_in_ls: + self._open_in_ls() + + def _open_in_ls(self) -> None: + """ + Open the file in the language server if it is not already open. + """ + if self._is_open_in_ls: + return + self._is_open_in_ls = True + self.language_server.server.notify.did_open_text_document( + { # ty: ignore[invalid-argument-type] # dict built from LSPConstants keys; shape matches the TypedDict + LSPConstants.TEXT_DOCUMENT: { + LSPConstants.URI: self.uri, + LSPConstants.LANGUAGE_ID: self.language_id, + LSPConstants.VERSION: 0, + LSPConstants.TEXT: self.contents, + } + } + ) + + def close(self) -> None: + if self._is_open_in_ls: + self.language_server.server.notify.did_close_text_document( + { # ty: ignore[invalid-argument-type] # dict built from LSPConstants keys; shape matches the TypedDict + LSPConstants.TEXT_DOCUMENT: { + LSPConstants.URI: self.uri, + } + } + ) + + def ensure_open_in_ls(self) -> None: + """Ensure that the file is opened in the language server.""" + self._open_in_ls() + + @property + def contents(self) -> str: + file_modified_date = self.abs_path.stat().st_mtime + + # if contents are cached, check if they are stale (file modification since last read) and invalidate if so + if self._contents is not None: + assert self._read_file_modified_date is not None + if file_modified_date > self._read_file_modified_date: + self._contents = None + + if self._contents is None: + self._read_file_modified_date = file_modified_date + self._contents = FileUtils.read_file(str(self.abs_path), self.encoding) + self._content_hash = None + + return self._contents + + @contents.setter + def contents(self, new_contents: str) -> None: + """ + Sets new contents for the file buffer (in-memory change only). + Persistence of the change to disk must be handled separately. + + :param new_contents: the new contents to set + """ + self._contents = new_contents + self._content_hash = None + + @property + def content_hash(self) -> str: + if self._content_hash is None: + self._content_hash = hashlib.md5(self.contents.encode(self.encoding)).hexdigest() + return self._content_hash + + def split_lines(self) -> list[str]: + """Splits the contents of the file into lines.""" + return self.contents.split("\n") + + +class SymbolBody(ToStringMixin): + """ + Representation of the body of a symbol, which allows the extraction of the symbol's text + from the lines of the file it is defined in. + + Instances that share the same lines buffer are memory-efficient, + using only 4 integers and a reference to the lines buffer from which the text can be extracted, + i.e. a core representation of only about 40 bytes per body. + """ + + def __init__(self, lines: list[str], start_line: int, start_col: int, end_line: int, end_col: int) -> None: + self._lines = lines + self._start_line = start_line + self._start_col = start_col + self._end_line = end_line + self._end_col = end_col + + def _tostring_excludes(self) -> list[str]: + return ["_lines"] + + def get_text(self) -> str: + # extract relevant lines + symbol_body = "\n".join(self._lines[self._start_line : self._end_line + 1]) + + # remove leading content from the first line + symbol_body = symbol_body[self._start_col :] + + # remove trailing content from the last line + last_line = self._lines[self._end_line] + trailing_length = len(last_line) - self._end_col + if trailing_length > 0: + symbol_body = symbol_body[: -(len(last_line) - self._end_col)] + + return symbol_body + + +class SymbolBodyFactory: + """ + A factory for the creation of SymbolBody instances from symbols dictionaries. + Instances created from the same factory instance are memory-efficient, as they share + the same lines buffer. + """ + + def __init__(self, file_buffer: LSPFileBuffer): + self._lines = file_buffer.split_lines() + + def create_symbol_body(self, symbol: UnifiedSymbolInformation) -> SymbolBody: + existing_body = symbol.get("body", None) + if existing_body and isinstance(existing_body, SymbolBody): + return existing_body + + assert "location" in symbol + start_line = symbol["location"]["range"]["start"]["line"] + end_line = symbol["location"]["range"]["end"]["line"] + start_col = symbol["location"]["range"]["start"]["character"] + end_col = symbol["location"]["range"]["end"]["character"] + return SymbolBody(self._lines, start_line, start_col, end_line, end_col) + + +class DocumentSymbols: + # IMPORTANT: Instances of this class are persisted in the high-level document symbol cache + + def __init__(self, root_symbols: list[ls_types.UnifiedSymbolInformation]): + self.root_symbols = root_symbols + self._all_symbols: list[ls_types.UnifiedSymbolInformation] | None = None + + def __getstate__(self) -> dict: + return getstate(DocumentSymbols, self, transient_properties=["_all_symbols"]) + + def iter_symbols(self) -> Iterator[ls_types.UnifiedSymbolInformation]: + """ + Iterate over all symbols in the document symbol tree. + Yields symbols in a depth-first manner. + """ + if self._all_symbols is not None: + yield from self._all_symbols + return + + def traverse(s: ls_types.UnifiedSymbolInformation) -> Iterator[ls_types.UnifiedSymbolInformation]: + yield s + for child in s.get("children", []): + yield from traverse(child) + + for root_symbol in self.root_symbols: + yield from traverse(root_symbol) + + def get_all_symbols_and_roots(self) -> tuple[list[ls_types.UnifiedSymbolInformation], list[ls_types.UnifiedSymbolInformation]]: + """ + This function returns all symbols in the document as a flat list and the root symbols. + It exists to facilitate migration from previous versions, where this was the return interface of + the LS method that obtained document symbols. + + :return: A tuple containing a list of all symbols in the document and a list of root symbols. + """ + if self._all_symbols is None: + self._all_symbols = list(self.iter_symbols()) + return self._all_symbols, self.root_symbols + + +class LanguageServerDependencyProvider(ABC): + """ + Prepares dependencies for a language server (if any), ultimately enabling the launch command to be constructed + and optionally providing environment variables that are necessary for the execution. + """ + + def __init__(self, custom_settings: SolidLSPSettings.CustomLSSettings, ls_resources_dir: str): + """ + :param custom_settings: the (user-provided) language server-specific settings + :param ls_resources_dir: the directory in which data for the language-server shall be stored + (this is already specific to the concrete language server, i.e. no further subdirectory is needed) + """ + self._custom_settings = custom_settings + self._ls_resources_dir = ls_resources_dir + + @abstractmethod + def create_launch_command(self) -> list[str]: + """ + Creates the launch command for this language server, potentially downloading and installing dependencies + beforehand. + + :return: the launch command as a list containing the executable and its arguments + """ + + def create_launch_command_env(self) -> dict[str, str]: + """ + Provides environment variables to be set when executing the launch command. + + This method is intended to be overridden by subclasses that need to set variables. + + :return: a mapping for variable names to values + """ + return {} + + +class LanguageServerDependencyProviderBaseCommand(LanguageServerDependencyProvider, ABC): + """ + Special case of a dependency provider, where the launch command is constructed from a base command. + + The user can configure aspects of the launch command in LS-specific settings: + + * ``ls_base_cmd`: overrides the base command + * ``ls_path``: overrides the path of the language server's core dependency (e.g. its executable or a JAR file), + from which the base command is formed, bypassing Serena's managed installation + * ``ls_args``: overrides the arguments that are added to the base command in order to form the launch command + * ``ls_extra_args``: additional arguments to append to the launch command + """ + + @abstractmethod + def _create_default_base_command(self) -> list[str]: + """ + Obtains the default base command for this language server, potentially downloading and installing dependencies + beforehand. + + Note: The user can override the base command, so this will only be run if no custom base command is provided. + + :return: the base command as a list containing the executable and its arguments + """ + + @abstractmethod + def _create_launch_command_from_base_command(self, base_command: list[str]) -> list[str]: + """ + Adds any additional arguments to the base command to create the final launch command. + + :param base_command: the base command + :return: the extended command + """ + + def create_launch_command(self) -> list[str]: + # obtain base command + base_command = self._custom_settings.get("ls_base_cmd", None) + if base_command is not None and not isinstance(base_command, list): + log.warning("The 'ls_base_cmd' setting should be a list of strings. Ignoring the provided value: %s", base_command) + base_command = None + if base_command is None: + ls_path = self._custom_settings.get("ls_path", None) + if ls_path is not None: + base_command = [ls_path] + else: + # default case: base command is constructed by the provider implementation + base_command = self._create_default_base_command() + + # create launch command from base command + ls_args = self._custom_settings.get("ls_args") + if ls_args is not None: + if not isinstance(ls_args, list): + log.warning("The 'ls_args' setting should be a list of strings. Ignoring the provided value: %s", ls_args) + ls_args = None + if ls_args is not None: + cmd = list(base_command) + ls_args + else: + cmd = self._create_launch_command_from_base_command(list(base_command)) + + # add user-provided extra arguments (if any) + ls_extra_args = self._custom_settings.get("ls_extra_args", []) + if ls_extra_args: + if not isinstance(ls_extra_args, list): + log.warning("The 'ls_extra_args' setting should be a list of strings. Ignoring the provided value: %s", ls_extra_args) + else: + cmd = cmd + ls_extra_args + + return cmd + + +class LanguageServerDependencyProviderSinglePath(LanguageServerDependencyProviderBaseCommand, ABC): + """ + Special case of a dependency provider, where there is a single core dependency which provides + the basis for the launch command. + + The core dependency's path can be overridden by the user in LS-specific settings (SerenaConfig) + via the key "ls_path". If the user provides the key, the specified path is used directly. + Otherwise, the provider implementation is called to get or install the core dependency. + + Note: The inheritance from the BaseCommand class serves to allow user overrides to be handled + centrally, but implementations of this class do not necessarily follow the principle that + the base command is strictly extended. + Yet incompatibility arises only if (a) the user overrides the base command with a command that + has arguments, and (b) the implementation of this class does not construct the launch command + by appending arguments to the core dependency's path. + """ + + @abstractmethod + def _get_or_install_core_dependency(self) -> str: + """ + Gets the language server's core path, potentially installing dependencies beforehand. + + :return: the core dependency's path (e.g. executable, jar, etc.) + """ + + @abstractmethod + def _create_launch_command(self, core_path: str) -> list[str]: + """ + :param core_path: path to the core dependency + :return: the launch command as a list containing the executable and its arguments + """ + + def _create_default_base_command(self) -> list[str]: + # We treat the core path as the only element of the base command, + # noting that the construction of the launch command from the base command + # is not necessarily to append arguments only. + core_path = self._get_or_install_core_dependency() + return [core_path] + + def _create_launch_command_from_base_command(self, base_command: list[str]) -> list[str]: + core_path = base_command[0] + cmd = self._create_launch_command(core_path) + if len(base_command) == 1: + # This is the regular case, where the base command consists only of the core + # dependency's path, and the launch command is constructed from it. + return cmd + else: + # In this case, the user has overridden the base command via LS-specific settings + # with a command that has arguments and therefore does not map cleanly to + # the single path assumption. However, in special cases where the full command + # was constructed by appending arguments to the core dependency's path, + # we simply assume that the provided base command can be substituted. + if cmd[0] == core_path: + return base_command + cmd[1:] + else: + raise ValueError("Language server base launch command with arguments unsupported") + + +class LanguageServerDependencyProviderUvx(LanguageServerDependencyProviderBaseCommand): + """ + Dependency provider for language servers distributed as a PyPI package, run on demand via ``uvx`` / ``uv x``. + + The pinned package version can be overridden by the user via the LS-specific setting given by + ``version_setting_key``. Alternatively, the LS-specific setting "ls_path" can be set to the path of an + already-installed language server executable, in which case it is launched directly, bypassing uv entirely. + """ + + DEFAULT_UVX_PYTHON_VERSION = "3.13" + + def __init__( + self, + custom_settings: "SolidLSPSettings.CustomLSSettings", + ls_resources_dir: str, + *, + package: str, + entrypoint: str, + default_version: str, + version_setting_key: str, + extra_args: Sequence[str] = (), + ): + """ + :param package: the PyPI package name (e.g. ``"pyright"``) + :param entrypoint: the console script provided by the package (e.g. ``"pyright-langserver"``) + :param default_version: the package version to pin unless overridden + :param version_setting_key: the LS-specific setting key through which the user can override the version + :param extra_args: arguments appended after the entrypoint (e.g. ``("--stdio",)``) + """ + super().__init__(custom_settings, ls_resources_dir) + self._package = package + self._entrypoint = entrypoint + self._default_version = default_version + self._version_setting_key = version_setting_key + self._extra_args = tuple(extra_args) + + @staticmethod + def _build_uvx_base_command( + package: str, + version: str, + entrypoint: str, + python_version: str = DEFAULT_UVX_PYTHON_VERSION, + ) -> list[str]: + """Build a command that runs a pinned PyPI package's console script on demand via ``uvx`` / ``uv x``. + + Resolution order: + 1. Prefer ``uvx`` (env var ``UVX`` or PATH lookup). + 2. Fall back to ``uv x`` if only ``uv`` is on PATH. + 3. Raise ``RuntimeError`` if neither is available. + + :param package: PyPI package name (e.g. ``"pyright"``). + :param version: Pinned package version. + :param entrypoint: Console script provided by the package (e.g. ``"pyright-langserver"``). + :param python_version: Python interpreter version passed via ``-p`` (uv will fetch it if missing). + """ + base_args = ["-p", python_version, "--from", f"{package}=={version}", entrypoint] + + uvx_path = os.environ.get("UVX") or shutil.which("uvx") + if uvx_path is not None: + return [uvx_path, *base_args] + + uv_path = shutil.which("uv") + if uv_path is not None: + return [uv_path, "x", *base_args] + + raise RuntimeError("Could not find 'uvx' or 'uv' in PATH. Install uv (https://docs.astral.sh/uv/).") + + def _create_default_base_command(self): + version = self._custom_settings.get(self._version_setting_key, self._default_version) + return self._build_uvx_base_command(self._package, version, self._entrypoint) + + def _create_launch_command_from_base_command(self, base_command: list[str]) -> list[str]: + return base_command + list(self._extra_args) + + +class SolidLanguageServer(ABC): + """ + High-level abstraction for language server interaction, which wraps the underlying low-level LSP interface + """ + + CACHE_FOLDER_NAME = "cache" + RAW_DOCUMENT_SYMBOLS_CACHE_VERSION = 1 + """ + global version identifier for raw symbol caches; an LS-specific version is defined separately and combined with this. + This should be incremented whenever there is a change in the way raw document symbols are stored. + If the result of a language server changes in a way that affects the raw document symbols, + the LS-specific version should be incremented instead. + """ + RAW_DOCUMENT_SYMBOL_CACHE_FILENAME = "raw_document_symbols.pkl" + RAW_DOCUMENT_SYMBOL_CACHE_FILENAME_LEGACY_FALLBACK = "document_symbols_cache_v23-06-25.pkl" + DOCUMENT_SYMBOL_CACHE_VERSION = 4 + DOCUMENT_SYMBOL_CACHE_FILENAME = "document_symbols.pkl" + + # Directories that should always be ignored regardless of language: + # VCS internals, virtual environments, caches, and serena's own data. + _ALWAYS_IGNORED_DIRS = frozenset( + { + ".git", + ".svn", + ".hg", + ".bzr", # VCS + ".venv", + ".env", # virtual environments + ".cache", + ".mypy_cache", + ".pytest_cache", + ".ruff_cache", # caches + ".tox", + ".nox", # test runners + ".idea", # IDE internals + ".serena", # serena's own data + ".vscode", # Doesn't contain symbols + } + ) + + # To be overridden and extended by subclasses + def is_ignored_dirname(self, dirname: str) -> bool: + """ + A language-specific condition for directories that should always be ignored. For example, venv + in Python and node_modules in JS/TS should be ignored always. + """ + return dirname in self._ALWAYS_IGNORED_DIRS + + @staticmethod + def _determine_log_level(line: str) -> int: + """ + Classify a stderr line from the language server to determine appropriate logging level. + + Language servers may emit informational messages to stderr that contain words like "error" + but are not actual errors. Subclasses can override this method to filter out known + false-positive patterns specific to their language server. + + :param line: The stderr line to classify + :return: A logging level (logging.DEBUG, logging.INFO, logging.WARNING, or logging.ERROR) + """ + line_lower = line.lower() + + # Default classification: treat lines with "error" or "exception" as ERROR level + if "error" in line_lower or "exception" in line_lower or line.startswith("E["): + return logging.ERROR + else: + return logging.INFO + + @classmethod + def get_language_enum_instance(cls) -> Language: + return Language.from_ls_class(cls) + + @classmethod + def supports_implementation_request(cls) -> bool: + """ + Return whether this language server supports ``textDocument/implementation``. + """ + return False + + @classmethod + def ls_resources_dir(cls, solidlsp_settings: SolidLSPSettings, mkdir: bool = True) -> str: + """ + Returns the directory where the language server resources are downloaded. + This is used to store language server binaries, configuration files, etc. + """ + result = os.path.join(solidlsp_settings.ls_resources_dir, cls.__name__) + + # Migration of previously downloaded LS resources that were downloaded to a subdir of solidlsp instead of to the user's home + pre_migration_ls_resources_dir = os.path.join(os.path.dirname(__file__), "language_servers", "static", cls.__name__) + if os.path.exists(pre_migration_ls_resources_dir): + if os.path.exists(result): + # if the directory already exists, we just remove the old resources + shutil.rmtree(result, ignore_errors=True) + else: + # move old resources to the new location + shutil.move(pre_migration_ls_resources_dir, result) + if mkdir: + os.makedirs(result, exist_ok=True) + return result + + @classmethod + def create( + cls, + config: LanguageServerConfig, + repository_root_path: str, + timeout: float | None = None, + solidlsp_settings: SolidLSPSettings | None = None, + ) -> "SolidLanguageServer": + """ + Creates a language specific LanguageServer instance based on the given configuration, and appropriate settings for the programming language. + + If language is Java, then ensure that jdk-17.0.6 or higher is installed, `java` is in PATH, and JAVA_HOME is set to the installation directory. + If language is JS/TS, then ensure that node (v18.16.0 or higher) is installed and in PATH. + + :param repository_root_path: The root path of the repository. + :param config: language server configuration. + :param logger: The logger to use. + :param timeout: the timeout for requests to the language server. If None, no timeout will be used. + :param solidlsp_settings: additional settings + :return LanguageServer: A language specific LanguageServer instance. + """ + ls: SolidLanguageServer + if solidlsp_settings is None: + solidlsp_settings = SolidLSPSettings() + + # Ensure repository_root_path is absolute to avoid issues with file URIs + repository_root_path = os.path.abspath(repository_root_path) + + ls_class = config.code_language.get_ls_class() + # All language server implementations are required to use the same signature of the constructor + # (which differs from the signature of the base class constructor). + ls = ls_class(config, repository_root_path, solidlsp_settings) + ls.set_request_timeout(timeout) + return ls + + def __init__( + self, + config: LanguageServerConfig, + repository_root_path: str, + process_launch_info: ProcessLaunchInfo | None, + language_id: str, + solidlsp_settings: SolidLSPSettings, + cache_version_raw_document_symbols: Hashable = 1, + ): + """ + Initializes a LanguageServer instance. + + Do not instantiate this class directly. Use `LanguageServer.create` method instead. + + :param config: the language server's configuration. + :param repository_root_path: the root path of the repository. + :param process_launch_info: (DEPRECATED: pass None and implement _create_dependency_provider instead) + the command used to start the actual language server. + The command must pass appropriate flags to the binary, so that it runs in the stdio mode, + as opposed to HTTP, TCP modes supported by some language servers. + :param language_id: The language identifier which will be passed to the language server in the `textDocument/didOpen` + notification by default. + If the language server uses multiple language identifiers, it must override the method `get_language_id_for_file` + to provide the appropriate identifier for each type of file. + :param solidlsp_settings: the global SolidLSP settings + :param cache_version_raw_document_symbols: the version, for caching, of the raw document symbols coming + from this specific language server. This should be incremented by subclasses calling this constructor + whenever the format of the raw document symbols changes (typically because the language server + improves/fixes its output). + """ + self.config = config + self._solidlsp_settings = solidlsp_settings + lang = self.get_language_enum_instance() + self._custom_settings = solidlsp_settings.get_ls_specific_settings(lang) + """ + the (user-provided) language server-specific settings + """ + self._ls_resources_dir = self.ls_resources_dir(solidlsp_settings) + log.debug(f"Custom config (LS-specific settings) for {lang}: {self._custom_settings}") + self._encoding = config.encoding + self.repository_root_path: str = repository_root_path + + log.debug( + f"Creating language server instance for {repository_root_path=} with {language_id=} and process launch info: {process_launch_info}" + ) + + self.language_id = language_id + """ + default language identifier to be passed to the language server in `textDocument/didOpen` notifications. + """ + self.open_file_buffers: dict[str, LSPFileBuffer] = {} + self.language = self.get_language_enum_instance() + """ + identifies the language server (not to be confused with the language id passed to the language server) + """ + # The source filename matcher is a @cache'd per-language singleton. A previous project may + # have extended it (e.g. Perl's file_filter adding .cgi); reset it here so every activation + # starts from the language's default extensions, then language-server subclasses re-apply + # their own settings during the rest of __init__. + self.language.get_source_fn_matcher().reset() + self._published_diagnostics: dict[str, list[ls_types.Diagnostic]] = {} + self._published_diagnostics_generation_by_uri: dict[str, int] = {} + self._published_diagnostics_generation = 0 + self._published_diagnostics_condition = threading.Condition() + + # initialise symbol caches + self.cache_dir = Path(self._solidlsp_settings.project_data_path) / self.CACHE_FOLDER_NAME / self.language_id + self.cache_dir.mkdir(parents=True, exist_ok=True) + # * raw document symbols cache + self._ls_specific_raw_document_symbols_cache_version = cache_version_raw_document_symbols + self._raw_document_symbols_cache: dict[str, tuple[str, list[DocumentSymbol] | list[SymbolInformation] | None]] = {} + """maps relative file paths to a tuple of (file_content_hash, raw_root_symbols)""" + self._raw_document_symbols_cache_is_modified: bool = False + self._load_raw_document_symbols_cache() + # * high-level document symbols cache + self._document_symbols_cache: dict[str, tuple[str, DocumentSymbols]] = {} + """maps relative file paths to a tuple of (file_content_hash, document_symbols)""" + self._document_symbols_cache_is_modified: bool = False + self._load_document_symbols_cache() + + self.server_started = False + if config.trace_lsp_communication: + + def logging_fn(source: str, target: str, msg: StringDict | str) -> None: + log.debug(f"LSP: {source} -> {target}: {msg!s}") + + else: + logging_fn = None + + # create the LanguageServerHandler, which provides the functionality to start the language server and communicate with it, + # preparing the launch command beforehand + self._dependency_provider: LanguageServerDependencyProvider | None = None + if process_launch_info is None: + 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 = 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 + # for all absolute paths in ignored_paths, convert them to relative paths + processed_patterns = [] + for pattern in set(config.ignored_paths): + # Normalize separators (pathspec expects forward slashes) + pattern = pattern.replace(os.path.sep, "/") + processed_patterns.append(pattern) + log.debug(f"Processing {len(processed_patterns)} ignored paths from the config") + + # Create a pathspec matcher from the processed patterns + self._ignore_spec = pathspec.PathSpec.from_lines(pathspec.patterns.GitWildMatchPattern, processed_patterns) + + self._has_waited_for_cross_file_references = False + + self._abs_workspace_folders_indexed = self.config.get_absolute_workspace_folders(self.repository_root_path) + self._abs_workspace_folders_additional = self.config.get_absolute_additional_workspace_folders(self.repository_root_path) + """ + additional workspace folders, which are passed to the language server in the initialization request + but which are not indexed by SolidLSP, i.e. not traversed for full symbol tree + """ + self._abs_workspace_folders_all = self._abs_workspace_folders_indexed + self._abs_workspace_folders_additional + + @property + def custom_settings(self) -> SolidLSPSettings.CustomLSSettings: + """ + The (user-provided) language server-specific settings. + """ + return self._custom_settings + + 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.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. + + This is used for generic cross-language bookkeeping that must work independently of + language-specific notification handlers. + """ + if method == "textDocument/publishDiagnostics": + self._store_published_diagnostics(params) + + def _store_published_diagnostics(self, params: Any) -> None: + """ + Store diagnostics received through ``textDocument/publishDiagnostics``. + """ + if not isinstance(params, dict): + return + + uri = params.get("uri") + diagnostics = params.get("diagnostics") + if not isinstance(uri, str) or not isinstance(diagnostics, list): + return + + normalized_diagnostics: list[ls_types.Diagnostic] = [] + for diagnostic in diagnostics: + if not isinstance(diagnostic, dict): + continue + if "message" not in diagnostic or "range" not in diagnostic: + continue + + normalized_diagnostic: ls_types.Diagnostic = { + "uri": uri, + "message": diagnostic["message"], + "range": diagnostic["range"], + } + severity = diagnostic.get("severity") + if isinstance(severity, int): + normalized_diagnostic["severity"] = ls_types.DiagnosticSeverity(severity) + + code = diagnostic.get("code") + if isinstance(code, int | str): + normalized_diagnostic["code"] = code + + if "source" in diagnostic: + normalized_diagnostic["source"] = diagnostic["source"] + normalized_diagnostics.append(ls_types.Diagnostic(**normalized_diagnostic)) + + # canonicalize the key so lookups using URIs produced by pathlib.Path.as_uri() match + # what servers publish (e.g. file:///c%3A/... or file:///c:/... vs. file:///C:/...) + key = self._canonicalize_published_diagnostics_uri(uri) + + with self._published_diagnostics_condition: + self._published_diagnostics_generation += 1 + self._published_diagnostics[key] = normalized_diagnostics + self._published_diagnostics_generation_by_uri[key] = self._published_diagnostics_generation + self._published_diagnostics_condition.notify_all() + + @staticmethod + def _canonicalize_published_diagnostics_uri(uri: str) -> str: + """ + Canonicalizes a ``file://`` URI so that diagnostics published by language servers + and lookups based on ``pathlib.Path.as_uri()`` agree on the same key. + + On Windows, servers may publish under ``file:///c%3A/...`` or ``file:///c:/...`` while + ``pathlib.Path.as_uri()`` produces ``file:///C:/...``. The canonical form uses an + upper-case drive letter and a plain colon. + """ + if os.name != "nt" or not uri.startswith("file:///"): + return uri + + # extract the segment after "file:///" up to the next slash and look for a drive letter + prefix = "file:///" + rest = uri[len(prefix) :] + slash = rest.find("/") + head = rest if slash < 0 else rest[:slash] + tail = "" if slash < 0 else rest[slash:] + + if (len(head) >= 2 and head[0].isalpha() and head[1] == ":") or ( + len(head) >= 4 and head[0].isalpha() and head[1:4].lower() == "%3a" + ): + head = head[0].upper() + ":" + else: + return uri + + return prefix + head + tail + + def _get_published_diagnostics_generation(self, uri: str) -> int: + key = self._canonicalize_published_diagnostics_uri(uri) + with self._published_diagnostics_condition: + return self._published_diagnostics_generation_by_uri.get(key, -1) + + def _wait_for_published_diagnostics( + self, + uri: str, + after_generation: int, + timeout: float, + ) -> list[ls_types.Diagnostic] | None: + key = self._canonicalize_published_diagnostics_uri(uri) + deadline = perf_counter() + timeout + with self._published_diagnostics_condition: + while True: + current_generation = self._published_diagnostics_generation_by_uri.get(key, -1) + if current_generation > after_generation: + return list(self._published_diagnostics.get(key, [])) + + remaining_timeout = deadline - perf_counter() + if remaining_timeout <= 0: + return None + self._published_diagnostics_condition.wait(timeout=remaining_timeout) + + def _get_cached_published_diagnostics(self, uri: str) -> list[ls_types.Diagnostic] | None: + key = self._canonicalize_published_diagnostics_uri(uri) + with self._published_diagnostics_condition: + diagnostics = self._published_diagnostics.get(key) + if diagnostics is None: + return None + return list(diagnostics) + + @staticmethod + def _diagnostic_matches_range(diagnostic: ls_types.Diagnostic, start_line: int, end_line: int) -> bool: + diagnostic_start_line = diagnostic["range"]["start"]["line"] + diagnostic_end_line = diagnostic["range"]["end"]["line"] + + # normalize inverted ranges: some servers (e.g. Julia's JuliaSyntax.jl for unterminated + # expressions) emit diagnostics whose end position lies before the start position. + # treat such ranges as a single span covering both lines. + lo = min(diagnostic_start_line, diagnostic_end_line) + hi = max(diagnostic_start_line, diagnostic_end_line) + + # when end_line < 0 the caller imposes no upper bound; only enforce the lower bound. + if end_line < 0: + return hi >= start_line + return lo <= end_line and hi >= start_line + + @staticmethod + def _diagnostic_matches_min_severity(diagnostic: ls_types.Diagnostic, min_severity: int) -> bool: + severity = diagnostic.get("severity") + if severity is None: + return True + return int(severity) <= min_severity + + @classmethod + def _filter_diagnostics( + cls, + diagnostics: list[ls_types.Diagnostic], + start_line: int, + end_line: int, + min_severity: int, + ) -> list[ls_types.Diagnostic]: + diagnostics = [d for d in diagnostics if cls._diagnostic_matches_range(d, start_line, end_line)] + diagnostics = [d for d in diagnostics if cls._diagnostic_matches_min_severity(d, min_severity)] + return diagnostics + + def _validate_text_document_diagnostics_request( + self, + relative_file_path: str, + start_line: int, + end_line: int, + min_severity: int, + ) -> str: + if not self.server_started: + log.error("request_text_document_diagnostics called before Language Server started") + raise SolidLSPException("Language Server not started") + if start_line < 0: + raise ValueError(f"start_line must be non-negative, got {start_line}") + if end_line != -1 and end_line < start_line: + raise ValueError(f"end_line must be -1 or >= start_line, got {end_line} < {start_line}") + if min_severity not in {1, 2, 3, 4}: + raise ValueError(f"min_severity must be one of 1, 2, 3, 4, got {min_severity}") + return pathlib.Path(str(PurePath(self.repository_root_path, relative_file_path))).as_uri() + + def get_published_diagnostics_generation(self, relative_file_path: str) -> int: + """ + Get the generation number for the latest published diagnostics of a file. + + :param relative_file_path: The relative path of the file. + :return: the generation number, or ``-1`` if none were published yet. + """ + uri = pathlib.Path(str(PurePath(self.repository_root_path, relative_file_path))).as_uri() + return self._get_published_diagnostics_generation(uri) + + def get_cached_published_text_document_diagnostics( + self, + relative_file_path: str, + start_line: int = 0, + end_line: int = -1, + min_severity: int = 4, + ) -> list[ls_types.Diagnostic] | None: + """ + Get cached diagnostics received through ``textDocument/publishDiagnostics``. + + :param relative_file_path: The relative path of the file to retrieve diagnostics for. + :param start_line: the first 0-based line to include in the result. + :param end_line: the last 0-based line to include in the result. ``-1`` means no upper bound. + :param min_severity: minimum LSP severity to include, where 1=Error, 2=Warning, 3=Information, 4=Hint. + Diagnostics with lower-or-equal numeric severity are returned. + :return: the cached diagnostics, or ``None`` if no diagnostics were published yet. + """ + uri = self._validate_text_document_diagnostics_request(relative_file_path, start_line, end_line, min_severity) + diagnostics = self._get_cached_published_diagnostics(uri) + if diagnostics is None: + return None + return self._filter_diagnostics(diagnostics, start_line, end_line, min_severity) + + def request_published_text_document_diagnostics( + self, + relative_file_path: str, + after_generation: int = -1, + timeout: float = 2.5, + start_line: int = 0, + end_line: int = -1, + min_severity: int = 4, + allow_cached: bool = True, + ) -> list[ls_types.Diagnostic] | None: + """ + Wait for diagnostics received through ``textDocument/publishDiagnostics`` and return them. + + :param relative_file_path: The relative path of the file to retrieve diagnostics for. + :param after_generation: only return diagnostics published after this generation. ``-1`` accepts the next publication. + :param timeout: the maximum time to wait for a newer publication. + :param start_line: the first 0-based line to include in the result. + :param end_line: the last 0-based line to include in the result. ``-1`` means no upper bound. + :param min_severity: minimum LSP severity to include, where 1=Error, 2=Warning, 3=Information, 4=Hint. + Diagnostics with lower-or-equal numeric severity are returned. + :param allow_cached: whether to fall back to the current cached diagnostics if no newer publication arrives in time. + :return: the published diagnostics, or ``None`` if no diagnostics are available. + """ + uri = self._validate_text_document_diagnostics_request(relative_file_path, start_line, end_line, min_severity) + published_uri = self._get_published_diagnostics_uri(uri) + diagnostics: list[ls_types.Diagnostic] | None = None + + # keeping the document open + with self.open_file(relative_file_path): + diagnostics = self._wait_for_relevant_published_diagnostics( + uri=published_uri, + after_generation=after_generation, + timeout=timeout, + allow_cached=allow_cached, + ) + + if diagnostics is None: + return None + + return self._filter_diagnostics(diagnostics, start_line, end_line, min_severity) + + def _get_published_diagnostics_uri(self, request_uri: str) -> str: + """ + Gets the URI under which published diagnostics should be looked up. + """ + return request_uri + + def _supports_pull_diagnostics(self) -> bool: + """ + Whether the language server handles ``textDocument/diagnostic`` (LSP 3.17 pull + diagnostics) gracefully. Subclasses should override and return ``False`` when the + underlying server does not implement pull diagnostics, especially when sending + the request would terminate the server (e.g. LanguageServer.jl raises an error + for unknown methods that crashes the process). + """ + return True + + def _get_published_diagnostics_wait_timeout(self, pull_diagnostics_failed: bool) -> float: + """ + Gets the timeout for waiting on published diagnostics after a diagnostics request. + """ + return 2.5 + + def _accept_published_diagnostics(self, diagnostics: list[ls_types.Diagnostic]) -> bool: + """ + Determines whether a published diagnostics payload should satisfy the current wait. + """ + return bool(diagnostics) + + def _wait_for_relevant_published_diagnostics( + self, + uri: str, + after_generation: int, + timeout: float, + allow_cached: bool = True, + ) -> list[ls_types.Diagnostic] | None: + """ + Waits for a published diagnostics payload that is relevant for the current request. + """ + deadline = monotonic() + timeout + current_after_generation = after_generation + + while True: + remaining_timeout = deadline - monotonic() + if remaining_timeout <= 0: + break + + diagnostics = self._wait_for_published_diagnostics( + uri=uri, + after_generation=current_after_generation, + timeout=remaining_timeout, + ) + if diagnostics is None: + break + if self._accept_published_diagnostics(diagnostics): + return diagnostics + current_after_generation = self._get_published_diagnostics_generation(uri) + + if allow_cached: + diagnostics = self._get_cached_published_diagnostics(uri) + if diagnostics is not None and self._accept_published_diagnostics(diagnostics): + return diagnostics + + return None + + def request_text_document_diagnostics( + self, + relative_file_path: str, + start_line: int = 0, + end_line: int = -1, + min_severity: int = 4, + ) -> list[ls_types.Diagnostic]: + """ + Raise a [textDocument/diagnostic](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#textDocument_diagnostic) request to the Language Server + to find diagnostics for the given file. Wait for the response and return the result. + + :param relative_file_path: The relative path of the file to retrieve diagnostics for + :param start_line: the first 0-based line to include in the result. + :param end_line: the last 0-based line to include in the result. `-1` means no upper bound. + :param min_severity: minimum LSP severity to include, where 1=Error, 2=Warning, 3=Information, 4=Hint. + Diagnostics with lower-or-equal numeric severity are returned. + + :return: A list of diagnostics for the file + """ + uri = self._validate_text_document_diagnostics_request(relative_file_path, start_line, end_line, min_severity) + published_uri = self._get_published_diagnostics_uri(uri) + diagnostics_before_request = self._get_published_diagnostics_generation(published_uri) + ret: list[ls_types.Diagnostic] | None = None + pull_diagnostics_failed = False + + with self.open_file(relative_file_path): + response: Any = None + # only send pull diagnostics when the server actually supports it; some servers + # (e.g. Julia's LanguageServer.jl) hard-error and crash the process on unknown methods + if self._supports_pull_diagnostics(): + try: + response = self.server.send.text_document_diagnostic( + { # ty: ignore[invalid-argument-type] # dict built from LSPConstants keys; shape matches the TypedDict + LSPConstants.TEXT_DOCUMENT: { + LSPConstants.URI: uri, + } + } + ) + except SolidLSPException as ex: + log.debug("Falling back to published diagnostics for %s due to pull-diagnostics error: %s", relative_file_path, ex) + response = None + pull_diagnostics_failed = True + + if response is not None: + assert isinstance(response, dict), ( + f"Unexpected response from Language Server (expected list, got {type(response)}): {response}" + ) + ret = [] + for item in response["items"]: # type: ignore + new_item: ls_types.Diagnostic = { + "uri": uri, + "severity": item["severity"], + "message": item["message"], + "range": item["range"], + "code": item.get("code"), # type: ignore + } + if "source" in item: + new_item["source"] = item["source"] + ret.append(ls_types.Diagnostic(**new_item)) + + if not ret: + published_diagnostics = self._wait_for_relevant_published_diagnostics( + uri=published_uri, + after_generation=diagnostics_before_request, + timeout=self._get_published_diagnostics_wait_timeout(pull_diagnostics_failed), + ) + if published_diagnostics is not None: + ret = published_diagnostics + + if ret is None: + return [] + + return self._filter_diagnostics(ret, start_line, end_line, min_severity) + + 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. + + LS may return incomplete results on calls to `request_references` (only references found in the same file), + if the LS is not fully initialized yet. + """ + return 2 + + # --- Cross-workspace / additional workspace folder support --- + + @staticmethod + def _path_contains_dots(relative_file_path: str) -> bool: + """Check if a relative path traverses outside the workspace root via '..' components.""" + return ".." in PurePath(relative_file_path).parts + + @dataclass + class PathWorkspaceStatus: + """ + Represents the status of a path with respect to the language server's workspace folders. + """ + + ls: "SolidLanguageServer" + resolve_abs_path: Path + is_in_workspace_folder: bool + workspace_root: str | None = None + """ + the workspace root the path is within/relative to, if any + """ + + @classmethod + def from_abs_resolved_path(cls, path: Path, ls: "SolidLanguageServer") -> "SolidLanguageServer.PathWorkspaceStatus": + """ + :param path: an absolute, resolved Path instance + :param ls: the language server instance + """ + for workspace in ls._abs_workspace_folders_all: + if path.is_relative_to(workspace): + return SolidLanguageServer.PathWorkspaceStatus( + ls=ls, is_in_workspace_folder=True, workspace_root=workspace, resolve_abs_path=path + ) + return SolidLanguageServer.PathWorkspaceStatus(ls=ls, is_in_workspace_folder=False, resolve_abs_path=path) + + @classmethod + def from_relative_path(cls, relative_path: str, ls: "SolidLanguageServer") -> "SolidLanguageServer.PathWorkspaceStatus": + """ + :param relative_path: a relative path from the repository root + :param ls: the language server instance + """ + return cls.from_abs_resolved_path(pathlib.Path(ls.repository_root_path, relative_path).resolve(), ls) + + def check_within_workspace_or_raise(self): + if not self.is_in_workspace_folder: + raise ValueError( + f"Path {self.resolve_abs_path} is outside of configured workspaces. " + f"Configured workspaces: {self.ls._abs_workspace_folders_all}." + ) + + def _resolve_file_uri(self, relative_file_path: str) -> str: + """Construct a canonical file URI from a relative path. + + For cross-workspace paths containing '..', the path is resolved to + produce a clean URI without '..' segments. + """ + p = pathlib.Path(os.path.join(self.repository_root_path, relative_file_path)) + if self._path_contains_dots(relative_file_path): + p = p.resolve() + self.PathWorkspaceStatus.from_abs_resolved_path(p, self).check_within_workspace_or_raise() + return p.as_uri() + + def _activate_additional_workspaces(self) -> None: + """Open a representative file from each additional workspace folder to + trigger project loading in the language server. + + Many language servers only load projects on-demand when files are opened. + This method finds a source file in each additional workspace folder via + :meth:`_find_representative_source_file` and opens it, keeping it open + for the lifetime of this language server instance. + + Subclasses must override :meth:`_find_representative_source_file` to + enable this feature; the default implementation raises ``NotImplementedError``. + """ + opened_count = 0 + for additional_workspace in self._abs_workspace_folders_additional: + source_file = self._find_representative_source_file(additional_workspace) + if source_file is None: + log.warning("No source file found in additional workspace folder: %s", additional_workspace) + continue + + rel_path = os.path.relpath(source_file, self.repository_root_path) + log.info("Opening %s to trigger project loading for %s", rel_path, os.path.basename(additional_workspace)) + + abs_path = pathlib.Path(source_file).resolve() + uri = abs_path.as_uri() + language_id = self._get_language_id_for_file(rel_path) + + self._signal_expect_indexing() + fb = LSPFileBuffer( + abs_path=abs_path, + uri=uri, + encoding=self._encoding, + version=0, + language_id=language_id, + ref_count=1, + language_server=self, + open_in_ls=True, + ) + self.open_file_buffers[uri] = fb + opened_count += 1 + + if opened_count > 0: + log.info("Waiting for %d additional workspace(s) to index...", opened_count) + self._wait_for_additional_workspace_indexing() + + def _find_representative_source_file(self, directory: str) -> str | None: + """Find a source file suitable for triggering project loading in the given directory. + + Must be overridden by subclasses that support ``additional_workspace_folders``. + Return ``None`` if no suitable file is found. + + :param directory: Absolute path to the workspace folder. + :return: Absolute path to a source file, or None. + """ + raise NotImplementedError( + f"{type(self).__name__} does not yet support additional_workspace_folders. " + f"Override _find_representative_source_file() to enable this feature." + ) + + def _signal_expect_indexing(self) -> None: + """Signal that new files are about to be opened and async indexing should be awaited. + + Override in subclasses that track indexing progress (e.g. via $/progress). + Default implementation is a no-op. + """ + + def _wait_for_additional_workspace_indexing(self) -> None: + """Wait for additional workspace indexing to complete. + + Override in subclasses that track indexing progress. + Default implementation is a no-op. + """ + + def set_request_timeout(self, timeout: float | None) -> None: + """ + :param timeout: the timeout, in seconds, for requests to the language server. + """ + self.server.set_request_timeout(timeout) + + def get_ignore_spec(self) -> pathspec.PathSpec: + """ + Returns the pathspec matcher for the paths that were configured to be ignored through + the language server configuration. + + This is a subset of the full language-specific ignore spec that determines + which files are relevant for the language server. + + This matcher is useful for operations outside of the language server, + such as when searching for relevant non-language files in the project. + """ + return self._ignore_spec + + def get_source_fn_matcher(self) -> FilenameMatcher: + """ + :return: the source filename matcher for this language server, which must positively match all files that + are understood by this language server or are discovered as containing sources indirectly, e.g. via references + """ + # By default, use the matcher of the language + return self.language.get_source_fn_matcher() + + def is_ignored_path(self, relative_path: str, ignore_unsupported_files: bool = True) -> bool: + """ + Determine if a path should be ignored based on file type + and ignore patterns. + + Missing paths are classified without raising, since language servers may report + locations of generated files that are not present on disk (e.g. compiled classes + under build output). A missing path is treated as a file if its name has a suffix + and as a directory otherwise; directory-only ignore patterns (e.g. ``build/``) + match only existing directories. + + :param relative_path: Relative path to check + :param ignore_unsupported_files: whether files that are not supported source files should be ignored + + :return: True if the path should be ignored, False otherwise + """ + abs_path = os.path.join(self.repository_root_path, relative_path) + rel_path = Path(relative_path) + + # determine whether the path is a file: from the filesystem if the path exists, lexically otherwise + exists = os.path.exists(abs_path) + if not exists: + log.debug("Path %s does not exist; the ignore check is performed on the path itself", abs_path) + is_file = os.path.isfile(abs_path) if exists else bool(rel_path.suffix) + + # check each directory component against always-ignored dirnames + dir_parts = rel_path.parts + if is_file: + dir_parts = dir_parts[:-1] + for part in dir_parts: + if not part: # Skip empty parts (e.g., from leading '/') + continue + if self.is_ignored_dirname(part): + return True + + # apply the unsupported-extension rule to files + if is_file and ignore_unsupported_files: + fn_matcher = self.get_source_fn_matcher() + if not fn_matcher.is_relevant_filename(abs_path): + return True + + return match_path(relative_path, self.get_ignore_spec(), root_path=self.repository_root_path) + + @contextmanager + def start_server_context(self) -> Iterator["SolidLanguageServer"]: + self.start() + try: + yield self + finally: + self.stop() + + @abstractmethod + def _start_server(self) -> None: + """ + 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: + """ + Determines the language identifier to pass to the language server for the given file, + particularly `textDocument/didOpen` requests. + + Override this method in subclasses to return file-specific language identifiers. + The default implementation returns the main identifier passed at construction (self.language_id). + """ + return self.language_id + + @contextmanager + def open_file(self, relative_file_path: str, open_in_ls: bool = True) -> Iterator[LSPFileBuffer]: + """ + Open a file in the Language Server. This is required before making any requests to the Language Server. + + :param relative_file_path: The relative path of the file to open. + :param open_in_ls: whether to open the file in the language server, sending the didOpen notification. + Set this to False to read the local file buffer without notifying the LS; the file can + be opened in the LS later by calling the `ensure_open_in_ls` method on the returned LSPFileBuffer. + """ + if not self.server_started: + log.error("open_file called before Language Server started") + raise SolidLSPException("Language Server not started") + + absolute_file_path = Path(self.repository_root_path, relative_file_path) + if self._path_contains_dots(relative_file_path): + absolute_file_path = absolute_file_path.resolve() + uri = absolute_file_path.as_uri() + + if uri in self.open_file_buffers: + fb = self.open_file_buffers[uri] + assert fb.uri == uri + assert fb.ref_count >= 1 + + fb.ref_count += 1 + if open_in_ls: + fb.ensure_open_in_ls() + else: + version = 0 + language_id = self._get_language_id_for_file(relative_file_path) + fb = LSPFileBuffer( + abs_path=absolute_file_path, + uri=uri, + encoding=self._encoding, + version=version, + language_id=language_id, + ref_count=1, + language_server=self, + open_in_ls=open_in_ls, + ) + self.open_file_buffers[uri] = fb + + try: + yield fb + finally: + fb.ref_count -= 1 + if fb.ref_count == 0: + fb.close() + del self.open_file_buffers[uri] + + @contextmanager + def _open_file_context( + self, relative_file_path: str, file_buffer: LSPFileBuffer | None = None, open_in_ls: bool = True + ) -> Iterator[LSPFileBuffer]: + """ + Internal context manager to open a file, optionally reusing an existing file buffer. + + :param relative_file_path: the relative path of the file to open. + :param file_buffer: an optional existing file buffer to reuse. + :param open_in_ls: whether to open the file in the language server, sending the didOpen notification. + Set this to False to read the local file buffer without notifying the LS; the file can + be opened in the LS later by calling the `ensure_open_in_ls` method on the returned LSPFileBuffer. + """ + if file_buffer is not None: + expected_uri = self._resolve_file_uri(relative_file_path) + assert file_buffer.uri == expected_uri, f"Inconsistency between provided {file_buffer.uri=} and {expected_uri=}" + if open_in_ls: + file_buffer.ensure_open_in_ls() + yield file_buffer + else: + with self.open_file(relative_file_path, open_in_ls=open_in_ls) as fb: + yield fb + + def insert_text_at_position(self, relative_file_path: str, line: int, column: int, text_to_be_inserted: str) -> ls_types.Position: + """ + Insert text at the given line and column in the given file and return + the updated cursor position after inserting the text. + + :param relative_file_path: The relative path of the file to open. + :param line: The line number at which text should be inserted. + :param column: The column number at which text should be inserted. + :param text_to_be_inserted: The text to insert. + """ + if not self.server_started: + log.error("insert_text_at_position called before Language Server started") + raise SolidLSPException("Language Server not started") + + uri = self._resolve_file_uri(relative_file_path) + + # Ensure the file is open + assert uri in self.open_file_buffers + + file_buffer = self.open_file_buffers[uri] + file_buffer.version += 1 + + new_contents, new_l, new_c = TextUtils.insert_text_at_position(file_buffer.contents, line, column, text_to_be_inserted) + file_buffer.contents = new_contents + self.server.notify.did_change_text_document( + { # ty: ignore[invalid-argument-type] # dict built from LSPConstants keys; shape matches the TypedDict + LSPConstants.TEXT_DOCUMENT: { + LSPConstants.VERSION: file_buffer.version, + LSPConstants.URI: file_buffer.uri, + }, + LSPConstants.CONTENT_CHANGES: [ + { + LSPConstants.RANGE: { + "start": {"line": line, "character": column}, + "end": {"line": line, "character": column}, + }, + "text": text_to_be_inserted, + } + ], + } + ) + return ls_types.Position(line=new_l, character=new_c) + + def delete_text_between_positions( + self, + relative_file_path: str, + start: ls_types.Position, + end: ls_types.Position, + ) -> str: + """ + Delete text between the given start and end positions in the given file and return the deleted text. + """ + if not self.server_started: + log.error("delete_text_between_positions called before Language Server started") + raise SolidLSPException("Language Server not started") + + uri = self._resolve_file_uri(relative_file_path) + + # Ensure the file is open + assert uri in self.open_file_buffers + + file_buffer = self.open_file_buffers[uri] + file_buffer.version += 1 + new_contents, deleted_text = TextUtils.delete_text_between_positions( + file_buffer.contents, start_line=start["line"], start_col=start["character"], end_line=end["line"], end_col=end["character"] + ) + file_buffer.contents = new_contents + self.server.notify.did_change_text_document( + { # ty: ignore[invalid-argument-type] # dict built from LSPConstants keys; shape matches the TypedDict + LSPConstants.TEXT_DOCUMENT: { + LSPConstants.VERSION: file_buffer.version, + LSPConstants.URI: file_buffer.uri, + }, + LSPConstants.CONTENT_CHANGES: [{LSPConstants.RANGE: {"start": start, "end": end}, "text": ""}], + } + ) + return deleted_text + + def _send_definition_request(self, definition_params: DefinitionParams) -> Definition | list[LocationLink] | None: + return self.server.send.definition(definition_params) + + class SymbolLocationRequest(ABC): + def __init__( + self, + language_server: "SolidLanguageServer", + relative_file_path: str, + line: int, + column: int, + *, + request_name: str, + ) -> None: + self.language_server = language_server + self.relative_file_path = relative_file_path + self.line = line + self.column = column + self.request_name = request_name + self.skip_ignored_paths = True + + def execute(self) -> list[ls_types.Location]: + self._ensure_server_started() + + t0 = perf_counter() if _debug_enabled else None + # arm indexing tracking before didOpen so that the subsequent wait can + # observe the indexing progress triggered by opening the file + self.language_server._pre_open_for_cross_file_references() + with self.language_server.open_file(self.relative_file_path): + self.language_server._wait_for_cross_file_references_if_needed() + try: + response = self.send_request() + except Exception as e: + mapped_exception = self.map_exception(e) + if mapped_exception is not None: + raise mapped_exception from e + raise + + result = self.normalize_response(response) + if t0 is not None: + self.log_perf_result(t0, result) + return result + + def _ensure_server_started(self) -> None: + if not self.language_server.server_started: + log.error("%s called before language server started", self.request_name) + raise SolidLSPException("Language Server not started") + + @abstractmethod + def send_request(self) -> object | None: + pass + + def map_exception(self, error: Exception) -> Exception | None: + if isinstance(error, LSPError) and getattr(error, "code", None) == -32603: + return RuntimeError( + f"LSP internal error (-32603) when requesting {self.request_name} for " + f"{self.relative_file_path}:{self.line}:{self.column}. " + "This often occurs when requesting a symbol in a way the language server cannot resolve." + ) + return None + + @abstractmethod + def normalize_response(self, response: object | None) -> list[ls_types.Location]: + pass + + def convert_location_item(self, item: dict[str, object], *, allow_location_links: bool = False) -> ls_types.Location | None: + if LSPConstants.URI in item and LSPConstants.RANGE in item: + uri = cast(str, item[LSPConstants.URI]) + range_d = cast(ls_types.Range, item[LSPConstants.RANGE]) + elif ( + allow_location_links + and LSPConstants.TARGET_URI in item + and LSPConstants.TARGET_RANGE in item + and LSPConstants.TARGET_SELECTION_RANGE in item + ): + uri = cast(str, item[LSPConstants.TARGET_URI]) + range_d = cast(ls_types.Range, item[LSPConstants.TARGET_SELECTION_RANGE]) + else: + raise AssertionError(f"Unexpected response from Language Server: {item}") + + abs_path = PathUtils.uri_to_path(uri) + rel_path_str = PathUtils.get_relative_path(abs_path, self.language_server.repository_root_path) + + if rel_path_str is None: + log.warning( + "Found a %s in a path outside the repository, probably the LS is parsing things in installed packages or in the standardlib! " + "Path: %s. This is a bug but we currently simply skip these locations.", + self.request_name, + abs_path, + ) + return None + + # skip locations of generated files that are not present on disk + # (language servers may report e.g. compiled classes under build output) + if not os.path.exists(abs_path): + log.info("%s found symbol at non-existent path: %s", self.request_name, abs_path) + return None + + if self.skip_ignored_paths and self.language_server.is_ignored_path(rel_path_str): + log.info("%s found symbol in ignored path: %s", self.request_name, rel_path_str) + return None + + return ls_types.Location(uri=uri, range=range_d, absolutePath=str(abs_path), relativePath=rel_path_str) + + def log_perf_result(self, t0: float, result: list[ls_types.Location]) -> None: + return + + class DefinitionLocationRequest(SymbolLocationRequest): + def __init__( + self, + language_server: "SolidLanguageServer", + relative_file_path: str, + line: int, + column: int, + *, + request_name: str = "request_definition", + ) -> None: + super().__init__( + language_server, + relative_file_path, + line, + column, + request_name=request_name, + ) + + def send_request(self) -> object | None: + return self.language_server._send_definition_request( + self.language_server._create_text_document_position_params(self.relative_file_path, self.line, self.column) + ) + + def normalize_response(self, response: object | None) -> list[ls_types.Location]: + if response is None: + log.warning( + "Language server returned None for %s request at %s:%s:%s", + self.request_name, + self.relative_file_path, + self.line, + self.column, + ) + return [] + + ret: list[ls_types.Location] = [] + if isinstance(response, list): + for item in response: + assert isinstance(item, dict), f"Unexpected response from Language Server (expected dict, got {type(item)}): {item}" + if location := self.convert_location_item(cast(dict[str, object], item), allow_location_links=True): + ret.append(location) + return ret + + if isinstance(response, dict): + if location := self.convert_location_item(cast(dict[str, object], response), allow_location_links=True): + ret.append(location) + return ret + + assert False, f"Unexpected response from Language Server: {response}" + + def request_definition(self, relative_file_path: str, line: int, column: int) -> list[ls_types.Location]: + """ + Raise a [textDocument/definition](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#textDocument_definition) request to the Language Server + for the symbol at the given line and column in the given file. Wait for the response and return the result. + + :param relative_file_path: The relative path of the file that has the symbol for which definition should be looked up + :param line: The line number of the symbol + :param column: The column number of the symbol + + :return: the list of locations where the symbol is defined + """ + request = self.DefinitionLocationRequest(self, relative_file_path, line, column) + return request.execute() + + def _send_implementation_request(self, implementation_params: ImplementationParams) -> Definition | list[LocationLink] | None: + return self.server.send.implementation(implementation_params) + + def _create_text_document_position_params(self, relative_file_path: str, line: int, column: int) -> DefinitionParams: + return cast( + DefinitionParams, + { + LSPConstants.TEXT_DOCUMENT: {LSPConstants.URI: self._resolve_file_uri(relative_file_path)}, + LSPConstants.POSITION: { + LSPConstants.LINE: line, + LSPConstants.CHARACTER: column, + }, + }, + ) + + def _pre_open_for_cross_file_references(self) -> None: + """Called just before open_file() for requests that may need cross-file indexing. + + Override to pre-arm async indexing tracking before sending didOpen (e.g. clear + a threading.Event so a subsequent wait for indexing blocks correctly). + The base implementation is a no-op. + """ + + def _wait_for_cross_file_references_if_needed(self) -> None: + if not self._has_waited_for_cross_file_references: + # Some LS require waiting for a while before they can return accurate cross-file results. + # The waiting has to happen after at least one file was opened in the LS. + sleep(self._get_wait_time_for_cross_file_referencing()) + self._has_waited_for_cross_file_references = True + + class ImplementationLocationRequest(DefinitionLocationRequest): + def __init__(self, language_server: "SolidLanguageServer", relative_file_path: str, line: int, column: int) -> None: + super().__init__( + language_server, + relative_file_path, + line, + column, + request_name="request_implementation", + ) + + def send_request(self) -> object | None: + return self.language_server._send_implementation_request( + self.language_server._create_text_document_position_params(self.relative_file_path, self.line, self.column), + ) + + def request_implementation(self, relative_file_path: str, line: int, column: int) -> list[ls_types.Location]: + """ + Raise a [textDocument/implementation](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#textDocument_implementation) + request to the Language Server for the symbol at the given line and column in the given file. + + :param relative_file_path: The relative path of the file that has the symbol for which implementations should be looked up + :param line: The line number of the symbol + :param column: The column number of the symbol + :return: the list of locations where the symbol is implemented + """ + request = self.ImplementationLocationRequest(self, relative_file_path, line, column) + return request.execute() + + # Some LS cause problems with this, so the call is isolated from the rest to allow overriding in subclasses + def _send_references_request(self, relative_file_path: str, line: int, column: int) -> list[lsp_types.Location] | None: + return self.server.send.references( + { + "textDocument": {"uri": self._resolve_file_uri(relative_file_path)}, + "position": {"line": line, "character": column}, + "context": {"includeDeclaration": False}, + } + ) + + class ReferencesLocationRequest(SymbolLocationRequest): + def __init__(self, language_server: "SolidLanguageServer", relative_file_path: str, line: int, column: int) -> None: + super().__init__( + language_server, + relative_file_path, + line, + column, + request_name="request_references", + ) + + def send_request(self) -> object | None: + return self.language_server._send_references_request(self.relative_file_path, line=self.line, column=self.column) + + def normalize_response(self, response: object | None) -> list[ls_types.Location]: + if response is None: + return [] + + assert isinstance(response, list), f"Unexpected response from Language Server (expected list, got {type(response)}): {response}" + ret: list[ls_types.Location] = [] + for item in response: + assert isinstance(item, dict), f"Unexpected response from Language Server (expected dict, got {type(item)}): {item}" + if location := self.convert_location_item(cast(dict[str, object], item)): + ret.append(location) + return ret + + def log_perf_result(self, t0: float, result: list[ls_types.Location]) -> None: + elapsed_ms = (perf_counter() - t0) * 1000 + if not result: + log.debug("perf: request_references path=%s elapsed_ms=%.2f count=0", self.relative_file_path, elapsed_ms) + return + + unique_files = len({r["relativePath"] for r in result}) + log.debug( + "perf: request_references path=%s elapsed_ms=%.2f count=%d unique_files=%d", + self.relative_file_path, + elapsed_ms, + len(result), + unique_files, + ) + + def request_references(self, relative_file_path: str, line: int, column: int) -> list[ls_types.Location]: + """ + Raise a [textDocument/references](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#textDocument_references) request to the Language Server + to find references to the symbol at the given line and column in the given file. Wait for the response and return the result. + Filters out references located in ignored directories. + + :param relative_file_path: The relative path of the file that has the symbol for which references should be looked up + :param line: The line number of the symbol + :param column: The column number of the symbol + + :return: A list of locations where the symbol is referenced (excluding ignored directories) + """ + request = self.ReferencesLocationRequest(self, relative_file_path, line, column) + return request.execute() + + def retrieve_full_file_content(self, file_path: str) -> str: + """ + Retrieve the full content of the given file. + """ + if os.path.isabs(file_path): + file_path = os.path.relpath(file_path, self.repository_root_path) + with self.open_file(file_path) as file_data: + return file_data.contents + + def retrieve_content_around_line( + self, relative_file_path: str, line: int, context_lines_before: int = 0, context_lines_after: int = 0 + ) -> MatchedConsecutiveLines: + """ + Retrieve the content of the given file around the given line. + + :param relative_file_path: The relative path of the file to retrieve the content from + :param line: The line number to retrieve the content around + :param context_lines_before: The number of lines to retrieve before the given line + :param context_lines_after: The number of lines to retrieve after the given line + + :return MatchedConsecutiveLines: A container with the desired lines. + """ + with self.open_file(relative_file_path) as file_data: + file_contents = file_data.contents + return MatchedConsecutiveLines.from_file_contents( + file_contents, + line=line, + context_lines_before=context_lines_before, + context_lines_after=context_lines_after, + source_file_path=relative_file_path, + ) + + def request_completions( + self, relative_file_path: str, line: int, column: int, allow_incomplete: bool = False + ) -> list[ls_types.CompletionItem]: + """ + Raise a [textDocument/completion](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#textDocument_completion) request to the Language Server + to find completions at the given line and column in the given file. Wait for the response and return the result. + + :param relative_file_path: The relative path of the file that has the symbol for which completions should be looked up + :param line: The line number of the symbol + :param column: The column number of the symbol + + :return: A list of completions + """ + with self.open_file(relative_file_path): + open_file_buffer = self.open_file_buffers[self._resolve_file_uri(relative_file_path)] + completion_params: LSPTypes.CompletionParams = { + "position": {"line": line, "character": column}, + "textDocument": {"uri": open_file_buffer.uri}, + "context": {"triggerKind": LSPTypes.CompletionTriggerKind.Invoked}, + } + response: list[LSPTypes.CompletionItem] | LSPTypes.CompletionList | None = None + + for _ in range(30): + response = self.server.send.completion(completion_params) + if isinstance(response, list): + response = {"items": response, "isIncomplete": False} + if response is None or not response["isIncomplete"]: + break + + # TODO: Understand how to appropriately handle `isIncomplete` + if response is None or (response["isIncomplete"] and not allow_incomplete): + return [] + + if "items" in response: + response = response["items"] + + response = cast(list[LSPTypes.CompletionItem], response) + + # TODO: Handle the case when the completion is a keyword + items = [item for item in response if item["kind"] != LSPTypes.CompletionItemKind.Keyword] + + completions_list: list[ls_types.CompletionItem] = [] + + for item in items: + assert "insertText" in item or "textEdit" in item + assert "kind" in item + completion_item = {} + if "detail" in item: + completion_item["detail"] = item["detail"] + + if "label" in item: + completion_item["completionText"] = item["label"] + completion_item["kind"] = item["kind"] + elif "insertText" in item: + completion_item["completionText"] = item["insertText"] + completion_item["kind"] = item["kind"] + elif "textEdit" in item and "newText" in item["textEdit"]: + completion_item["completionText"] = item["textEdit"]["newText"] + completion_item["kind"] = item["kind"] + elif "textEdit" in item and "range" in item["textEdit"]: + new_dot_lineno, new_dot_colno = ( + completion_params["position"]["line"], + completion_params["position"]["character"], + ) + assert all( + ( + item["textEdit"]["range"]["start"]["line"] == new_dot_lineno, + item["textEdit"]["range"]["start"]["character"] == new_dot_colno, + item["textEdit"]["range"]["start"]["line"] == item["textEdit"]["range"]["end"]["line"], + item["textEdit"]["range"]["start"]["character"] == item["textEdit"]["range"]["end"]["character"], + ) + ) + + completion_item["completionText"] = item["textEdit"]["newText"] + completion_item["kind"] = item["kind"] + elif "textEdit" in item and "insert" in item["textEdit"]: + assert False + else: + assert False + + completion_item = ls_types.CompletionItem(**completion_item) # type: ignore + completions_list.append(completion_item) + + return [json.loads(json_repr) for json_repr in set(json.dumps(item, sort_keys=True) for item in completions_list)] + + def _request_document_symbols( + self, relative_file_path: str, file_data: LSPFileBuffer | None + ) -> list[SymbolInformation] | list[DocumentSymbol] | None: + """ + Sends a [documentSymbol](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#textDocument_documentSymbol) + request to the language server to find symbols in the given file - or returns a cached result if available. + The returned symbols are considered "raw document symbols" (in contrast to processed symbols returned by `request_document_symbols`). + + NOTE: This method can be overridden in subclasses to post-process the raw results. + When doing so after the initial implementation, be sure to update the init parameter `cache_version_raw_document_symbols` + to a different version (add 1) to ensure that all caches are invalidated appropriately. + IMPORTANT: Since rebuilding the raw document symbol cache from the language server results + is potentially expensive, prefer overriding the `request_document_symbols` method + if the post-processing can also be done on the processed/high-level symbols. + + :param relative_file_path: the relative path of the file that has the symbols. + :param file_data: the file data buffer, if already opened. If None, the file will be opened in this method. + :return: the list of root symbols in the file. + """ + + def get_cached_raw_document_symbols(cache_key: str, fd: LSPFileBuffer) -> list[SymbolInformation] | list[DocumentSymbol] | None: + file_hash_and_result = self._raw_document_symbols_cache.get(cache_key) + if file_hash_and_result is None: + log.debug("No cache hit for raw document symbols in %s", relative_file_path) + log.debug("perf: raw_document_symbols_cache MISS path=%s", relative_file_path) + return None + + file_hash, result = file_hash_and_result + if file_hash == fd.content_hash: + log.debug("Returning cached raw document symbols for %s", relative_file_path) + log.debug("perf: raw_document_symbols_cache HIT path=%s", relative_file_path) + return result + + log.debug("Document content for %s has changed (raw symbol cache is not up-to-date)", relative_file_path) + log.debug("perf: raw_document_symbols_cache STALE path=%s", relative_file_path) + return None + + def get_raw_document_symbols(fd: LSPFileBuffer) -> list[SymbolInformation] | list[DocumentSymbol] | None: + # check for cached result + cache_key = relative_file_path + response = get_cached_raw_document_symbols(cache_key, fd) + if response is not None: + return response + + # no cached result, query language server + log.debug(f"Requesting document symbols for {relative_file_path} from the Language Server") + response = self.server.send.document_symbol({"textDocument": {"uri": self._resolve_file_uri(relative_file_path)}}) + + # Only cache non-empty results. An empty or None response can occur when the language server + # has not yet finished indexing or building the project (e.g. Lean 4 before `lake build`), + # and caching it would permanently serve stale data even after the project is ready. + if response: + self._raw_document_symbols_cache[cache_key] = (fd.content_hash, response) + self._raw_document_symbols_cache_is_modified = True + + return response + + with self._open_file_context(relative_file_path, file_buffer=file_data) as fd: + return get_raw_document_symbols(fd) + + def _normalize_symbol_name(self, symbol: RawDocumentSymbol, relative_file_path: str) -> str: + """ + Normalizes the name of the given symbol, e.g. by removing parameter lists from method symbols. + + Override this method in subclasses to implement language-specific normalization logic. + NOTE: When changing the override of this method after the initial LS implementation, + be sure to also override `_document_symbols_cache_fingerprint` in order to ensure that + the caches are invalidated appropriately. + + :param symbol: the symbol + :param relative_file_path: the relative path of the file the symbol is located in + :return: the normalized name of the symbol + """ + # the default implementation does not change the name + return symbol["name"] + + def request_document_symbols(self, relative_file_path: str, file_buffer: LSPFileBuffer | None = None) -> DocumentSymbols: + """ + Retrieves the collection of symbols in the given file. + + NOTE: This method can be overridden in subclasses to post-process the results. + When doing so after the initial LS implementation, be sure to also override `_document_symbols_cache_fingerprint` + to ensure that the caches are invalidated appropriately. + DO NOT override this method to modify symbol names; override `_normalize_symbol_name` instead. + + :param relative_file_path: The relative path of the file that has the symbols + :param file_buffer: an optional file buffer if the file is already opened. + :return: the collection of symbols in the file. + All contained symbols will have a location, children, and a parent attribute, + where the parent attribute is None for root symbols. + Note that this is slightly different from the call to request_full_symbol_tree, + where the parent attribute will be the file symbol which in turn may have a package symbol as parent. + If you need a symbol tree that contains file symbols as well, you should use `request_full_symbol_tree` instead. + """ + with self._open_file_context(relative_file_path, file_buffer, open_in_ls=False) as file_data: + # check if the desired result is cached + cache_key = relative_file_path + file_hash_and_result = self._document_symbols_cache.get(cache_key) + if file_hash_and_result is None: + log.debug("No cache hit for document symbols in %s", relative_file_path) + log.debug("perf: document_symbols_cache MISS path=%s", relative_file_path) + else: + file_hash, document_symbols = file_hash_and_result + if file_hash == file_data.content_hash: + log.debug("Returning cached document symbols for %s", relative_file_path) + log.debug("perf: document_symbols_cache HIT path=%s", relative_file_path) + return document_symbols + + log.debug("Cached document symbol content for %s has changed", relative_file_path) + log.debug("perf: document_symbols_cache STALE path=%s", relative_file_path) + + # no cached result: request the root symbols from the language server + root_symbols = self._request_document_symbols(relative_file_path, file_data) + + if root_symbols is None: + log.warning( + f"Received None response from the Language Server for document symbols in {relative_file_path}. " + f"This means the language server can't understand this file (possibly due to syntax errors). It may also be due to a bug or misconfiguration of the LS. " + f"Returning empty list", + ) + return DocumentSymbols([]) + + assert isinstance(root_symbols, list), f"Unexpected response from Language Server: {root_symbols}" + log.debug("Received %d root symbols for %s from the language server", len(root_symbols), relative_file_path) + + body_factory = SymbolBodyFactory(file_data) + + def convert_to_unified_symbol(original_symbol_dict: RawDocumentSymbol) -> ls_types.UnifiedSymbolInformation: + """ + Converts the given symbol dictionary to the unified representation, ensuring + that all required fields are present (except 'children' which is handled separately). + + :param original_symbol_dict: the item to augment + :return: the augmented item (new object) + """ + # noinspection PyInvalidCast + item = cast(ls_types.UnifiedSymbolInformation, dict(original_symbol_dict)) + absolute_path = os.path.join(self.repository_root_path, relative_file_path) + + # handle missing location and path entries + if "location" not in item: + uri = pathlib.Path(absolute_path).as_uri() + assert "range" in item + tree_location = ls_types.Location( + uri=uri, + range=item["range"], + absolutePath=absolute_path, + relativePath=relative_file_path, + ) + item["location"] = tree_location + location = item["location"] + if "absolutePath" not in location: + location["absolutePath"] = absolute_path + if "relativePath" not in location: + location["relativePath"] = relative_file_path + + item["body"] = self.create_symbol_body(item, factory=body_factory) + + # handle missing selectionRange + if "selectionRange" not in item: + if "range" in item: + item["selectionRange"] = item["range"] + else: + item["selectionRange"] = item["location"]["range"] + + return item + + def convert_symbols_with_common_parent( + symbols: list[DocumentSymbol] | list[SymbolInformation], + parent: ls_types.UnifiedSymbolInformation | None, + ) -> list[ls_types.UnifiedSymbolInformation]: + """ + Converts the given symbols into UnifiedSymbolInformation with proper parent-child relationships, + adding overload indices for symbols with the same name under the same parent. + """ + # apply name normalization and count occurrences of each symbol name + total_name_counts: dict[str, int] = defaultdict(lambda: 0) + for symbol in symbols: + name = self._normalize_symbol_name(symbol, relative_file_path=relative_file_path) + symbol["name"] = name + total_name_counts[name] += 1 + + # convert symbols to the unified representation and + # * add overload indices where necessary + # * ensure that the "parent" field is set correctly + name_counts: dict[str, int] = defaultdict(lambda: 0) + unified_symbols = [] + for symbol in symbols: + usymbol = convert_to_unified_symbol(symbol) + if total_name_counts[usymbol["name"]] > 1: + usymbol["overload_idx"] = name_counts[usymbol["name"]] + name_counts[usymbol["name"]] += 1 + usymbol["parent"] = parent + if "children" in usymbol: + usymbol["children"] = convert_symbols_with_common_parent(usymbol["children"], usymbol) # type: ignore + else: + usymbol["children"] = [] + unified_symbols.append(usymbol) + return unified_symbols + + unified_root_symbols = convert_symbols_with_common_parent(root_symbols, None) + document_symbols = DocumentSymbols(unified_root_symbols) + + # update cache + log.debug("Updating cached document symbols for %s", relative_file_path) + self._document_symbols_cache[cache_key] = (file_data.content_hash, document_symbols) + self._document_symbols_cache_is_modified = True + + return document_symbols + + def request_full_symbol_tree(self, within_relative_path: str | None = None) -> list[ls_types.UnifiedSymbolInformation]: + """ + Will go through all files in the project or within a relative path and build a tree of symbols. + Note: this may be slow the first time it is called, especially if `within_relative_path` is not used to restrict the search. + + For each file, a symbol of kind File (2) will be created. For directories, a symbol of kind Package (4) will be created. + All symbols will have a children attribute, thereby representing the tree structure of all symbols in the project + that are within the repository. + All symbols except the root packages will have a parent attribute. + Will ignore directories starting with '.', language-specific defaults + and user-configured directories (e.g. from .gitignore). + + :param within_relative_path: pass a relative path to only consider symbols within this path. + If a file is passed, only the symbols within this file will be considered. + If a directory is passed, all files within this directory will be considered. + :return: A list of root symbols representing the top-level packages/modules in the project. + """ + + # Helper function to recursively process directories + def process_directory(abs_dir_path: str) -> list[ls_types.UnifiedSymbolInformation]: + abs_dir_path = os.path.realpath(abs_dir_path) + + rel_dir_path: str | None + try: + rel_dir_path = str(Path(abs_dir_path).relative_to(self.repository_root_path)) + except ValueError: # not relative to repository root + rel_dir_path = None + if rel_dir_path and self.is_ignored_path(rel_dir_path): + log.debug("Skipping directory: %s (because it should be ignored)", rel_dir_path) + return [] + + result = [] + try: + contained_dir_or_file_names = os.listdir(abs_dir_path) + except OSError: + return [] + + # Create package symbol for directory + package_symbol = ls_types.UnifiedSymbolInformation( + name=os.path.basename(abs_dir_path), + kind=ls_types.SymbolKind.Package, + location=ls_types.Location( + uri=str(pathlib.Path(abs_dir_path).as_uri()), + range={"start": {"line": 0, "character": 0}, "end": {"line": 0, "character": 0}}, + absolutePath=str(abs_dir_path), + relativePath=str(Path(abs_dir_path).resolve().relative_to(self.repository_root_path)), + ), + children=[], + ) + result.append(package_symbol) + + for contained_dir_or_file_name in contained_dir_or_file_names: + contained_dir_or_file_abs_path = os.path.join(abs_dir_path, contained_dir_or_file_name) + + # obtain relative path + try: + contained_dir_or_file_rel_path = str( + Path(contained_dir_or_file_abs_path).resolve().relative_to(self.repository_root_path) + ) + except ValueError as e: + # Typically happens when the path is not under the repository root (e.g., symlink pointing outside) + log.warning( + "Skipping path %s; likely outside of the repository root %s [cause: %s]", + contained_dir_or_file_abs_path, + self.repository_root_path, + e, + ) + continue + + if self.is_ignored_path(contained_dir_or_file_rel_path): + log.debug("Skipping item: %s (because it should be ignored)", contained_dir_or_file_rel_path) + continue + + if os.path.isdir(contained_dir_or_file_abs_path): + child_symbols = process_directory(contained_dir_or_file_abs_path) + package_symbol["children"].extend(child_symbols) + for child in child_symbols: + child["parent"] = package_symbol + + elif os.path.isfile(contained_dir_or_file_abs_path): + with self._open_file_context(contained_dir_or_file_rel_path, open_in_ls=False) as file_data: + document_symbols = self.request_document_symbols(contained_dir_or_file_rel_path, file_data) + file_root_nodes = document_symbols.root_symbols + + # Create file symbol, link with children + file_range = self._get_range_from_file_content(file_data.contents) + file_symbol = ls_types.UnifiedSymbolInformation( + name=os.path.splitext(contained_dir_or_file_name)[0], + kind=ls_types.SymbolKind.File, + range=file_range, + selectionRange=file_range, + location=ls_types.Location( + uri=str(pathlib.Path(contained_dir_or_file_abs_path).as_uri()), + range=file_range, + absolutePath=str(contained_dir_or_file_abs_path), + relativePath=str(Path(contained_dir_or_file_abs_path).resolve().relative_to(self.repository_root_path)), + ), + children=file_root_nodes, + parent=package_symbol, + ) + for child in file_root_nodes: + child["parent"] = file_symbol + + # Link file symbol with package + package_symbol["children"].append(file_symbol) + + # TODO: Not sure if this is actually still needed given recent changes to relative path handling + def fix_relative_path(nodes: list[ls_types.UnifiedSymbolInformation]) -> None: + for node in nodes: + if "location" in node and "relativePath" in node["location"]: + path = Path(node["location"]["relativePath"]) # type: ignore + if path.is_absolute(): + try: + path = path.relative_to(self.repository_root_path) + node["location"]["relativePath"] = str(path) + except Exception: + pass + if "children" in node: + fix_relative_path(node["children"]) + + fix_relative_path(file_root_nodes) + + return result + + if within_relative_path: + within_abs_path = os.path.join(self.repository_root_path, within_relative_path) + if not os.path.exists(within_abs_path): + raise FileNotFoundError(f"File or directory not found: {within_abs_path}") + if self.is_ignored_path(within_relative_path): + raise ValueError(f"Explicitly requested symbols in '{within_relative_path}' while the path is ignored") + if os.path.isfile(within_abs_path): + root_nodes = self.request_document_symbols(within_relative_path).root_symbols + return root_nodes + else: + self.PathWorkspaceStatus.from_abs_resolved_path(Path(within_abs_path).resolve(), self).check_within_workspace_or_raise() + return process_directory(within_abs_path) + else: + full_result = [] + for root in self.config.get_absolute_workspace_folders(self.repository_root_path): + full_result.extend(process_directory(root)) + return full_result + + @staticmethod + def _get_range_from_file_content(file_content: str) -> ls_types.Range: + """ + Get the range for the given file. + """ + lines = file_content.split("\n") + end_line = len(lines) + end_column = len(lines[-1]) + return ls_types.Range(start=ls_types.Position(line=0, character=0), end=ls_types.Position(line=end_line, character=end_column)) + + def request_dir_overview(self, relative_dir_path: str) -> dict[str, list[UnifiedSymbolInformation]]: + """ + :return: A mapping of all relative paths analyzed to lists of top-level symbols in the corresponding file. + """ + symbol_tree = self.request_full_symbol_tree(relative_dir_path) + # Initialize result dictionary + result: dict[str, list[UnifiedSymbolInformation]] = defaultdict(list) + + # Helper function to process a symbol and its children + def process_symbol(symbol: ls_types.UnifiedSymbolInformation) -> None: + if symbol["kind"] == ls_types.SymbolKind.File: + # For file symbols, process their children (top-level symbols) + for child in symbol["children"]: + # Handle cross-platform path resolution (fixes Docker/macOS path issues) + absolute_path = Path(child["location"]["absolutePath"]).resolve() + repository_root = Path(self.repository_root_path).resolve() + + # Try pathlib first, fallback to alternative approach if paths are incompatible + try: + path = absolute_path.relative_to(repository_root) + except ValueError: + # If paths are from different roots (e.g., /workspaces vs /Users), + # use the relativePath from location if available, or extract from absolutePath + if "relativePath" in child["location"] and child["location"]["relativePath"]: + path = Path(child["location"]["relativePath"]) + else: + # Extract relative path by finding common structure + # Example: /workspaces/.../test_repo/file.py -> test_repo/file.py + path_parts = absolute_path.parts + + # Find the last common part or use a fallback + if "test_repo" in path_parts: + test_repo_idx = path_parts.index("test_repo") + path = Path(*path_parts[test_repo_idx:]) + else: + # Last resort: use filename only + path = Path(absolute_path.name) + result[str(path)].append(child) + # For package/directory symbols, process their children + for child in symbol["children"]: + process_symbol(child) + + # Process each root symbol + for root in symbol_tree: + process_symbol(root) + return result + + def request_document_overview(self, relative_file_path: str) -> list[UnifiedSymbolInformation]: + """ + :return: the top-level symbols in the given file. + """ + return self.request_document_symbols(relative_file_path).root_symbols + + def request_overview(self, within_relative_path: str) -> dict[str, list[UnifiedSymbolInformation]]: + """ + An overview of all symbols in the given file or directory. + Raises a ValueError if a path to an ignored file is passed. + + :param within_relative_path: the relative path to the file or directory to get the overview of. + :return: A mapping of all relative paths analyzed to lists of top-level symbols in the corresponding file. + """ + abs_path = (Path(self.repository_root_path) / within_relative_path).resolve() + if not abs_path.exists(): + raise FileNotFoundError(f"File or directory not found: {abs_path}") + + if abs_path.is_file(): + if self.is_ignored_path(within_relative_path): + raise ValueError(f"The explicitly passed file {within_relative_path} is ignored, not returning overview.") + symbols_overview = self.request_document_overview(within_relative_path) + return {within_relative_path: symbols_overview} + else: + return self.request_dir_overview(within_relative_path) + + def request_hover( + self, relative_file_path: str, line: int, column: int, file_buffer: LSPFileBuffer | None = None + ) -> ls_types.Hover | None: + """ + Raise a [textDocument/hover](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#textDocument_hover) request to the Language Server + to find the hover information at the given line and column in the given file. Wait for the response and return the result. + + :param relative_file_path: The relative path of the file that has the hover information + :param line: The line number of the symbol + :param column: The column number of the symbol + :param file_buffer: The file buffer to use for the request. If not provided, the file will be read from disk. + Can be used for optimizing number of file reads in downstream code + """ + with self._open_file_context(relative_file_path, file_buffer=file_buffer) as fb: + return self._request_hover(fb, line, column) + + def _request_hover(self, file_buffer: LSPFileBuffer, line: int, column: int) -> ls_types.Hover | None: + """ + Performs the actual hover request. + """ + response = self.server.send.hover( + { + "textDocument": {"uri": file_buffer.uri}, + "position": { + "line": line, + "character": column, + }, + } + ) + + if response is None: + return None + + assert isinstance(response, dict) + contents = response.get("contents") + if not contents: + return None + if isinstance(contents, dict) and not contents.get("value"): + return None + return ls_types.Hover(**response) # type: ignore + + def request_signature_help(self, relative_file_path: str, line: int, column: int) -> ls_types.SignatureHelp | None: + """ + Raise a [textDocument/signatureHelp](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#textDocument_signatureHelp) + request to the Language Server to find the signature help at the given line and column in the given file. + Note: contrary to `hover`, this only returns something on the position of a *call* and not on a symbol definition. + This means for Serena's purposes, this method is not particularly useful. The result is also fairly verbose (but well structured). + + :param relative_file_path: The relative path of the file that has the signature help + :param line: The line number of the symbol + :param column: The column number of the symbol + + :return None + """ + with self.open_file(relative_file_path): + response = self.server.send.signature_help( + { + "textDocument": {"uri": self._resolve_file_uri(relative_file_path)}, + "position": { + "line": line, + "character": column, + }, + } + ) + + if response is None: + return None + + assert isinstance(response, dict) + + return ls_types.SignatureHelp(**response) # type: ignore + + def create_symbol_body( + self, + symbol: ls_types.UnifiedSymbolInformation, + factory: SymbolBodyFactory | None = None, + ) -> SymbolBody: + if factory is None: + assert "relativePath" in symbol["location"] + with self._open_file_context(symbol["location"]["relativePath"]) as f: # type: ignore + factory = SymbolBodyFactory(f) + + return factory.create_symbol_body(symbol) + + def request_referencing_symbols( + self, + relative_file_path: str, + line: int, + column: int, + include_imports: bool = True, + include_self: bool = False, + include_body: bool = False, + include_file_symbols: bool = False, + ) -> list[ReferenceInSymbol]: + """ + Finds all symbols that reference the symbol at the given location. + This is similar to request_references but filters to only include symbols + (functions, methods, classes, etc.) that reference the target symbol. + + :param relative_file_path: The relative path to the file. + :param line: The 0-indexed line number. + :param column: The 0-indexed column number. + :param include_imports: whether to also include imports as references. + Unfortunately, the LSP does not have an import type, so the references corresponding to imports + will not be easily distinguishable from definitions. + :param include_self: whether to include the references that is the "input symbol" itself. + Only has an effect if the relative_file_path, line and column point to a symbol, for example a definition. + :param include_body: whether to include the body of the symbols in the result. + :param include_file_symbols: whether to include references that are file symbols. This + is often a fallback mechanism for when the reference cannot be resolved to a symbol. + :return: List of objects containing the symbol and the location of the reference. + """ + if not self.server_started: + log.error("request_referencing_symbols called before Language Server started") + raise SolidLSPException("Language Server not started") + + # First, get all references to the symbol + references = self.request_references(relative_file_path, line, column) + if not references: + return [] + + debug_enabled = log.isEnabledFor(logging.DEBUG) + t0_loop = perf_counter() if debug_enabled else 0.0 + # For each reference, find the containing symbol + result = [] + incoming_symbol = None + for ref in references: + ref_path = ref["relativePath"] + assert ref_path is not None + ref_line = ref["range"]["start"]["line"] + ref_col = ref["range"]["start"]["character"] + + with self.open_file(ref_path) as file_data: + body_factory = SymbolBodyFactory(file_data) + + # Get the containing symbol for this reference + containing_symbol = self.request_containing_symbol( + ref_path, ref_line, ref_col, include_body=include_body, body_factory=body_factory + ) + if containing_symbol is None: + # TODO: HORRIBLE HACK! I don't know how to do it better for now... + # THIS IS BOUND TO BREAK IN MANY CASES! IT IS ALSO SPECIFIC TO PYTHON! + # Background: + # When a variable is used to change something, like + # + # instance = MyClass() + # instance.status = "new status" + # + # we can't find the containing symbol for the reference to `status` + # since there is no container on the line of the reference + # The hack is to try to find a variable symbol in the containing module + # by using the text of the reference to find the variable name (In a very heuristic way) + # and then look for a symbol with that name and kind Variable + ref_text = file_data.contents.split("\n")[ref_line] + if "." in ref_text: + containing_symbol_name = ref_text.split(".")[0] + document_symbols = self.request_document_symbols(ref_path) + for symbol in document_symbols.iter_symbols(): + if symbol["name"] == containing_symbol_name and symbol["kind"] == ls_types.SymbolKind.Variable: + containing_symbol = copy(symbol) + containing_symbol["location"] = ref + containing_symbol["range"] = ref["range"] + break + + # We failed retrieving the symbol, falling back to creating a file symbol + if containing_symbol is None and include_file_symbols: + log.warning(f"Could not find containing symbol for {ref_path}:{ref_line}:{ref_col}. Returning file symbol instead") + fileRange = self._get_range_from_file_content(file_data.contents) + ref_abs_path = os.path.join(self.repository_root_path, ref_path) + if self._path_contains_dots(ref_path): + ref_abs_path = str(pathlib.Path(ref_abs_path).resolve()) + location = ls_types.Location( + uri=self._resolve_file_uri(ref_path), + range=fileRange, + absolutePath=ref_abs_path, + relativePath=ref_path, + ) + name = os.path.splitext(os.path.basename(ref_path))[0] + + containing_symbol = ls_types.UnifiedSymbolInformation( + kind=ls_types.SymbolKind.File, + range=fileRange, + selectionRange=fileRange, + location=location, + name=name, + children=[], + ) + + if include_body: + containing_symbol["body"] = self.create_symbol_body(containing_symbol, factory=body_factory) + + if containing_symbol is None or (not include_file_symbols and containing_symbol["kind"] == ls_types.SymbolKind.File): + continue + + assert "location" in containing_symbol + assert "selectionRange" in containing_symbol + + # Checking for self-reference + if ( + containing_symbol["location"]["relativePath"] == relative_file_path + and containing_symbol["selectionRange"]["start"]["line"] == ref_line + and containing_symbol["selectionRange"]["start"]["character"] == ref_col + ): + incoming_symbol = containing_symbol + if include_self: + result.append(ReferenceInSymbol(symbol=containing_symbol, line=ref_line, character=ref_col)) + continue + log.debug(f"Found self-reference for {incoming_symbol['name']}, skipping it since {include_self=}") + continue + + # checking whether reference is an import + # This is neither really safe nor elegant, but if we don't do it, + # there is no way to distinguish between definitions and imports as import is not a symbol-type + # and we get the type referenced symbol resulting from imports... + if ( + not include_imports + and incoming_symbol is not None + and containing_symbol["name"] == incoming_symbol["name"] + and containing_symbol["kind"] == incoming_symbol["kind"] + ): + log.debug( + f"Found import of referenced symbol {incoming_symbol['name']}" + f"in {containing_symbol['location']['relativePath']}, skipping" + ) + continue + + result.append(ReferenceInSymbol(symbol=containing_symbol, line=ref_line, character=ref_col)) + + if debug_enabled: + loop_elapsed_ms = (perf_counter() - t0_loop) * 1000 + unique_files = len({r.symbol["location"]["relativePath"] for r in result}) + log.debug( + "perf: request_referencing_symbols path=%s loop_elapsed_ms=%.2f ref_count=%d result_count=%d unique_files=%d", + relative_file_path, + loop_elapsed_ms, + len(references), + len(result), + unique_files, + ) + + return result + + def request_containing_symbol( + self, + relative_file_path: str, + line: int, + column: int | None = None, + strict: bool = False, + include_body: bool = False, + body_factory: SymbolBodyFactory | None = None, + ) -> ls_types.UnifiedSymbolInformation | None: + """ + Finds the first symbol containing the position for the given file. + For Python, container symbols are considered to be those with kinds corresponding to + functions, methods, or classes (typically: Function (12), Method (6), Class (5)). + + The method operates as follows: + - Request the document symbols for the file. + - Filter symbols to those that start at or before the given line. + - From these, first look for symbols whose range contains the (line, column). + - If one or more symbols contain the position, return the one with the greatest starting position + (i.e. the innermost container). + - If none (strictly) contain the position, return the symbol with the greatest starting position + among those above the given line. + - If no container candidates are found, return None. + + :param relative_file_path: The relative path to the Python file. + :param line: The 0-indexed line number. + :param column: The 0-indexed column (also called character). If not passed, the lookup will be based + only on the line. + :param strict: If True, the position must be strictly within the range of the symbol. + Setting to True is useful for example for finding the parent of a symbol, as with strict=False, + and the line pointing to a symbol itself, the containing symbol will be the symbol itself + (and not the parent). + :param include_body: Whether to include the body of the symbol in the result. + :return: The container symbol (if found) or None. + """ + # checking if the line is empty, unfortunately ugly and duplicating code, but I don't want to refactor + with self.open_file(relative_file_path): + absolute_file_path = os.path.join(self.repository_root_path, relative_file_path) + if self._path_contains_dots(relative_file_path): + absolute_file_path = str(pathlib.Path(absolute_file_path).resolve()) + content = FileUtils.read_file(absolute_file_path, self._encoding) + if content.split("\n")[line].strip() == "": + log.error(f"Passing empty lines to request_container_symbol is currently not supported, {relative_file_path=}, {line=}") + return None + + document_symbols = self.request_document_symbols(relative_file_path) + + # make jedi and pyright api compatible + # the former has no location, the later has no range + # we will just always add location of the desired format to all symbols + for symbol in document_symbols.iter_symbols(): + if "location" not in symbol: + range = symbol["range"] + location = ls_types.Location( + uri=f"file:/{absolute_file_path}", + range=range, + absolutePath=absolute_file_path, + relativePath=relative_file_path, + ) + symbol["location"] = location + else: + location = symbol["location"] + assert "range" in location + location["absolutePath"] = absolute_file_path + location["relativePath"] = relative_file_path + location["uri"] = Path(absolute_file_path).as_uri() + + # Allowed container kinds, currently only for Python + container_symbol_kinds = {ls_types.SymbolKind.Method, ls_types.SymbolKind.Function, ls_types.SymbolKind.Class} + + def is_position_in_range(line: int, range_d: ls_types.Range) -> bool: + start = range_d["start"] + end = range_d["end"] + + column_condition = True + if strict: + line_condition = end["line"] >= line > start["line"] + if column is not None and line == start["line"]: + column_condition = column > start["character"] + else: + line_condition = end["line"] >= line >= start["line"] + if column is not None and line == start["line"]: + column_condition = column >= start["character"] + return line_condition and column_condition + + # Only consider containers that are not one-liners (otherwise we may get imports) + candidate_containers = [ + s + for s in document_symbols.iter_symbols() + if s["kind"] in container_symbol_kinds and s["location"]["range"]["start"]["line"] != s["location"]["range"]["end"]["line"] + ] + var_containers = [s for s in document_symbols.iter_symbols() if s["kind"] == ls_types.SymbolKind.Variable] + candidate_containers.extend(var_containers) + + if not candidate_containers: + return None + + # From the candidates, find those whose range contains the given position. + containing_symbols = [] + for symbol in candidate_containers: + s_range = symbol["location"]["range"] + if not is_position_in_range(line, s_range): + continue + containing_symbols.append(symbol) + + if containing_symbols: + # Return the one with the greatest starting position (i.e. the innermost container). + containing_symbol = max(containing_symbols, key=lambda s: s["location"]["range"]["start"]["line"]) + if include_body: + containing_symbol["body"] = self.create_symbol_body(containing_symbol, factory=body_factory) + return containing_symbol + else: + return None + + def request_container_of_symbol( + self, symbol: ls_types.UnifiedSymbolInformation, include_body: bool = False + ) -> ls_types.UnifiedSymbolInformation | None: + """ + Finds the container of the given symbol if there is one. If the parent attribute is present, the parent is returned + without further searching. + + :param symbol: The symbol to find the container of. + :param include_body: whether to include the body of the symbol in the result. + :return: The container of the given symbol or None if no container is found. + """ + if "parent" in symbol: + return symbol["parent"] + assert "location" in symbol, f"Symbol {symbol} has no location and no parent attribute" + return self.request_containing_symbol( + symbol["location"]["relativePath"], # type: ignore + symbol["location"]["range"]["start"]["line"], + symbol["location"]["range"]["start"]["character"], + strict=True, + include_body=include_body, + ) + + def _get_preferred_definition(self, definitions: list[ls_types.Location]) -> ls_types.Location: + """ + Select the preferred definition from a list of definitions. + + When multiple definitions are returned (e.g., both source and type definitions), + this method determines which one to use. The base implementation simply returns + the first definition. + + Subclasses can override this method to implement language-specific preferences. + For example, TypeScript/Vue servers may prefer source files over .d.ts type + definition files. + + :param definitions: A non-empty list of definition locations. + :return: The preferred definition location. + """ + return definitions[0] + + def _get_document_symbols_with_locations(self, relative_file_path: str) -> list[ls_types.UnifiedSymbolInformation]: + abs_path = os.path.join(self.repository_root_path, relative_file_path) + if self._path_contains_dots(relative_file_path): + abs_path = str(pathlib.Path(abs_path).resolve()) + document_symbols = self.request_document_symbols(relative_file_path) + symbols = list(document_symbols.iter_symbols()) + + # Make SymbolInformation and DocumentSymbol shapes consistent by ensuring every + # symbol exposes a normalized location/range in the current workspace. + for symbol in symbols: + location = symbol["location"] + location["absolutePath"] = abs_path + location["relativePath"] = relative_file_path + location["uri"] = self._resolve_file_uri(relative_file_path) + return symbols + + @staticmethod + def _position_matches_range(range_d: ls_types.Range, line: int, column: int | None = None) -> bool: + start = range_d["start"] + end = range_d["end"] + if not (start["line"] <= line <= end["line"]): + return False + if column is None: + return True + if line == start["line"] and column < start["character"]: + return False + if line == end["line"] and column > end["character"]: + return False + return True + + @staticmethod + def _symbol_match_sort_key(symbol: ls_types.UnifiedSymbolInformation, match_priority: int) -> tuple[int, int, int, int, int]: + location = symbol["location"] + symbol_range = location["range"] + start = symbol_range["start"] + end = symbol_range["end"] + line_span = end["line"] - start["line"] + character_span = end["character"] - start["character"] if line_span == 0 else end["character"] + return match_priority, line_span, character_span, start["line"], start["character"] + + def _request_symbol_at_location( + self, + relative_file_path: str, + line: int, + column: int, + include_body: bool = False, + body_factory: SymbolBodyFactory | None = None, + ) -> ls_types.UnifiedSymbolInformation | None: + candidates: list[tuple[tuple[int, int, int, int, int], ls_types.UnifiedSymbolInformation]] = [] + for symbol in self._get_document_symbols_with_locations(relative_file_path): + location = symbol["location"] + symbol_range = location["range"] + selection_range = symbol.get("selectionRange") or symbol_range + + match_priority: int | None = None + if self._position_matches_range(selection_range, line, column): + match_priority = 0 + elif self._position_matches_range(symbol_range, line, column): + match_priority = 1 + else: + selection_start = selection_range["start"] + symbol_start = symbol_range["start"] + if (selection_start["line"], selection_start["character"]) == (line, column): + match_priority = 2 + elif (symbol_start["line"], symbol_start["character"]) == (line, column): + match_priority = 3 + elif selection_start["line"] == line and column <= selection_start["character"]: + match_priority = 4 + elif symbol_start["line"] == line and column <= symbol_start["character"]: + match_priority = 5 + + if match_priority is None: + continue + candidates.append((self._symbol_match_sort_key(symbol, match_priority), symbol)) + + if not candidates: + return None + + candidates.sort(key=lambda item: item[0]) + best_symbol = candidates[0][1] + if include_body: + best_symbol["body"] = self.create_symbol_body(best_symbol, factory=body_factory) + return best_symbol + + @staticmethod + def _iter_symbol_descendants(symbol: ls_types.UnifiedSymbolInformation) -> Iterator[ls_types.UnifiedSymbolInformation]: + """Yield descendant symbols in depth-first order.""" + for child in symbol.get("children", []): + yield child + yield from SolidLanguageServer._iter_symbol_descendants(child) + + def _refine_implementing_symbol( + self, + target_symbol: ls_types.UnifiedSymbolInformation | None, + implementing_symbol: ls_types.UnifiedSymbolInformation, + include_body: bool = False, + ) -> ls_types.UnifiedSymbolInformation: + """Resolve member-level implementation symbols when the LS returns a containing type.""" + if target_symbol is None: + return implementing_symbol + + target_kind = target_symbol["kind"] + if target_kind not in (ls_types.SymbolKind.Method, ls_types.SymbolKind.Function): + return implementing_symbol + + if implementing_symbol["kind"] == target_kind and implementing_symbol.get("name") == target_symbol.get("name"): + return implementing_symbol + + candidate_descendants: list[ls_types.UnifiedSymbolInformation] = [] + for descendant in self._iter_symbol_descendants(implementing_symbol): + if descendant.get("name") != target_symbol.get("name"): + continue + if descendant["kind"] != target_kind: + continue + candidate_descendants.append(descendant) + + if not candidate_descendants: + return implementing_symbol + + refined_symbol = min( + candidate_descendants, + key=lambda symbol: self._symbol_match_sort_key(symbol, match_priority=0), + ) + if include_body: + refined_symbol["body"] = self.create_symbol_body(refined_symbol) + return refined_symbol + + def request_symbol_at_location( + self, + relative_file_path: str, + line: int, + column: int, + include_body: bool = False, + ) -> ls_types.UnifiedSymbolInformation | None: + """ + Finds the symbol at the given position, preferring exact identifier matches and otherwise + falling back to the innermost symbol whose body contains the position. + + :param relative_file_path: The relative path to the file. + :param line: The 0-indexed line number. + :param column: The 0-indexed column number. + :param include_body: whether to include the body of the symbol in the result. + :return: The symbol at the given location, or None if no symbol could be resolved. + """ + if not self.server_started: + log.error("request_symbol_at_location called before language server started") + raise SolidLSPException("Language Server not started") + return self._request_symbol_at_location(relative_file_path, line, column, include_body=include_body) + + def request_defining_symbol( + self, + relative_file_path: str, + line: int, + column: int, + include_body: bool = False, + ) -> ls_types.UnifiedSymbolInformation | None: + """ + Finds the symbol that defines the symbol at the given location. + + This method first finds the definition of the symbol at the given position, + then retrieves the full symbol information for that definition. + + :param relative_file_path: The relative path to the file. + :param line: The 0-indexed line number. + :param column: The 0-indexed column number. + :param include_body: whether to include the body of the symbol in the result. + :return: The symbol information for the definition, or None if not found. + """ + if not self.server_started: + log.error("request_defining_symbol called before language server started") + raise SolidLSPException("Language Server not started") + + # Get the definition location(s) + definitions = self.request_definition(relative_file_path, line, column) + if not definitions: + return None + + # Select the preferred definition (subclasses can override _get_preferred_definition) + definition = self._get_preferred_definition(definitions) + def_path = definition["relativePath"] + if def_path is None: + return None + def_line = definition["range"]["start"]["line"] + def_col = definition["range"]["start"]["character"] + + return self._request_symbol_at_location( + def_path, + def_line, + def_col, + include_body=include_body, + ) + + def request_implementing_symbols( + self, + relative_file_path: str, + line: int, + column: int, + include_body: bool = False, + ) -> list[ls_types.UnifiedSymbolInformation]: + """ + Finds the symbols that implement the symbol at the given location. + + This method first finds implementation locations for the symbol at the given position, + then retrieves the full symbol information for each implementation and de-duplicates + results that map to the same containing symbol. + + :param relative_file_path: The relative path to the file. + :param line: The 0-indexed line number. + :param column: The 0-indexed column number. + :param include_body: whether to include the body of the symbols in the result. + :return: The symbol information for each implementation. + """ + if not self.server_started: + log.error("request_implementing_symbols called before language server started") + raise SolidLSPException("Language Server not started") + + target_symbol = self._request_symbol_at_location(relative_file_path, line, column, include_body=False) + implementation_locations = self.request_implementation(relative_file_path, line, column) + if not implementation_locations: + return [] + + result: list[ls_types.UnifiedSymbolInformation] = [] + seen_keys: set[tuple[str, int, int, int]] = set() + for implementation in implementation_locations: + implementation_path = implementation["relativePath"] + assert implementation_path is not None + implementation_line = implementation["range"]["start"]["line"] + implementation_col = implementation["range"]["start"]["character"] + implementing_symbol = self._request_symbol_at_location( + implementation_path, + implementation_line, + implementation_col, + include_body=include_body, + body_factory=None, + ) + if implementing_symbol is None: + continue + implementing_symbol = self._refine_implementing_symbol(target_symbol, implementing_symbol, include_body=include_body) + if "location" not in implementing_symbol: + continue + symbol_location = implementing_symbol["location"] + symbol_key = ( + cast(str, symbol_location["relativePath"]), + symbol_location["range"]["start"]["line"], + symbol_location["range"]["start"]["character"], + implementing_symbol["kind"], + ) + if symbol_key in seen_keys: + continue + seen_keys.add(symbol_key) + result.append(implementing_symbol) + + return result + + def _document_symbols_cache_fingerprint(self) -> Hashable | None: + """ + Returns a fingerprint of any language server-specific aspects that result in changes + to the high-level document symbol information. + + Language servers must implement this method/change the return value + * whenever they change the `request_document_symbols` implementation to modify the returned content + * are reconfigured in a way that affects the returned contents (e.g. context-specific configuration + such as build flags or environment variables); configuration options can, in such cases, be + hashed together to produce a single fingerprint value. + + Whenever the value changes, the document symbols cache will be invalidated and re-populated. + + The value must be hashable and safe for inclusion in cache version tuples. + E.g. use an integer, a string or a tuple of integers/strings. + + For example, if there is a single aspect being considered, use an integer to reflect the version + of this aspect (incrementing it whenever the implementation changes). + If multiple versioned aspects exist, use a tuple of versions, etc. + """ + return None + + def _document_symbols_cache_version(self) -> Hashable: + """ + Return the version for the document symbols cache. + + Incorporates cache context fingerprint if provided by the language server. + """ + fingerprint = self._document_symbols_cache_fingerprint() + if fingerprint is not None: + return (self.DOCUMENT_SYMBOL_CACHE_VERSION, fingerprint) + return self.DOCUMENT_SYMBOL_CACHE_VERSION + + def _save_raw_document_symbols_cache(self) -> None: + cache_file = self.cache_dir / self.RAW_DOCUMENT_SYMBOL_CACHE_FILENAME + + if not self._raw_document_symbols_cache_is_modified: + log.debug("No changes to raw document symbols cache, skipping save") + return + + log.info("Saving updated raw document symbols cache to %s", cache_file) + try: + save_cache(str(cache_file), self._raw_document_symbols_cache_version(), self._raw_document_symbols_cache) + self._raw_document_symbols_cache_is_modified = False + except Exception as e: + log.error( + "Failed to save raw document symbols cache to %s: %s. Note: this may have resulted in a corrupted cache file.", + cache_file, + e, + ) + + def _raw_document_symbols_cache_version(self) -> tuple[Hashable, ...]: + base_version: tuple[Hashable, ...] = (self.RAW_DOCUMENT_SYMBOLS_CACHE_VERSION, self._ls_specific_raw_document_symbols_cache_version) + fingerprint = self._document_symbols_cache_fingerprint() + if fingerprint is not None: + return (*base_version, fingerprint) + return base_version + + def _load_raw_document_symbols_cache(self) -> None: + cache_file = self.cache_dir / self.RAW_DOCUMENT_SYMBOL_CACHE_FILENAME + + if not cache_file.exists(): + # check for legacy cache to load to migrate + legacy_cache_file = self.cache_dir / self.RAW_DOCUMENT_SYMBOL_CACHE_FILENAME_LEGACY_FALLBACK + if legacy_cache_file.exists(): + try: + legacy_cache: dict[ + str, tuple[str, tuple[list[ls_types.UnifiedSymbolInformation], list[ls_types.UnifiedSymbolInformation]]] + ] = load_pickle(legacy_cache_file) + log.info("Migrating legacy document symbols cache with %d entries", len(legacy_cache)) + num_symbols_migrated = 0 + migrated_cache = {} + for cache_key, (file_hash, (all_symbols, root_symbols)) in legacy_cache.items(): + if cache_key.endswith("-True"): # include_body=True + new_cache_key = cache_key[:-5] + migrated_cache[new_cache_key] = (file_hash, root_symbols) + num_symbols_migrated += len(all_symbols) + log.info("Migrated %d document symbols from legacy cache", num_symbols_migrated) + self._raw_document_symbols_cache = migrated_cache + self._raw_document_symbols_cache_is_modified = True + self._save_raw_document_symbols_cache() + legacy_cache_file.unlink() + return + except Exception as e: + log.error("Error during cache migration: %s", e) + return + + # load existing cache (if any) + if cache_file.exists(): + log.info("Loading document symbols cache from %s", cache_file) + try: + saved_cache = load_cache(str(cache_file), self._raw_document_symbols_cache_version()) + if saved_cache is not None: + self._raw_document_symbols_cache = saved_cache + log.info(f"Loaded {len(self._raw_document_symbols_cache)} entries from raw document symbols cache.") + except Exception as e: + # cache can become corrupt, so just skip loading it + log.warning( + "Failed to load raw document symbols cache from %s (%s); Ignoring cache.", + cache_file, + e, + ) + + def _save_document_symbols_cache(self) -> None: + cache_file = self.cache_dir / self.DOCUMENT_SYMBOL_CACHE_FILENAME + + if not self._document_symbols_cache_is_modified: + log.debug("No changes to document symbols cache, skipping save") + return + + log.info("Saving updated document symbols cache to %s", cache_file) + try: + save_cache(str(cache_file), self._document_symbols_cache_version(), self._document_symbols_cache) + self._document_symbols_cache_is_modified = False + except Exception as e: + log.error( + "Failed to save document symbols cache to %s: %s. Note: this may have resulted in a corrupted cache file.", + cache_file, + e, + ) + + def _load_document_symbols_cache(self) -> None: + cache_file = self.cache_dir / self.DOCUMENT_SYMBOL_CACHE_FILENAME + if cache_file.exists(): + log.info("Loading document symbols cache from %s", cache_file) + try: + saved_cache = load_cache(str(cache_file), self._document_symbols_cache_version()) + if saved_cache is not None: + self._document_symbols_cache = saved_cache + log.info(f"Loaded {len(self._document_symbols_cache)} entries from document symbols cache.") + except Exception as e: + # cache can become corrupt, so just skip loading it + log.warning( + "Failed to load document symbols cache from %s (%s); Ignoring cache.", + cache_file, + e, + ) + + def save_cache(self) -> None: + self._save_raw_document_symbols_cache() + self._save_document_symbols_cache() + + def request_workspace_symbol(self, query: str) -> list[ls_types.UnifiedSymbolInformation] | None: + """ + Raise a [workspace/symbol](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#workspace_symbol) request to the Language Server + to find symbols across the whole workspace. Wait for the response and return the result. + + :param query: The query string to filter symbols by + + :return: A list of matching symbols + """ + response = self.server.send.workspace_symbol({"query": query}) + if response is None: + return None + + assert isinstance(response, list) + + ret: list[ls_types.UnifiedSymbolInformation] = [] + for item in response: + assert isinstance(item, dict) + + assert LSPConstants.NAME in item + assert LSPConstants.KIND in item + assert LSPConstants.LOCATION in item + + ret.append(ls_types.UnifiedSymbolInformation(**item)) # type: ignore + + return ret + + def request_rename_symbol_edit( + self, + relative_file_path: str, + line: int, + column: int, + new_name: str, + ) -> ls_types.WorkspaceEdit | None: + """ + Retrieve a WorkspaceEdit for renaming the symbol at the given location to the new name. + Does not apply the edit, just retrieves it. In order to actually rename the symbol, call apply_workspace_edit. + + :param relative_file_path: The relative path to the file containing the symbol + :param line: The 0-indexed line number of the symbol + :param column: The 0-indexed column number of the symbol + :param new_name: The new name for the symbol + :return: A WorkspaceEdit containing the changes needed to rename the symbol, or None if rename is not supported + """ + params = RenameParams( + textDocument=ls_types.TextDocumentIdentifier(uri=self._resolve_file_uri(relative_file_path)), + position=ls_types.Position(line=line, character=column), + newName=new_name, + ) + + with self.open_file(relative_file_path): + return self.server.send.rename(params) + + def apply_text_edits_to_file(self, relative_path: str, edits: list[ls_types.TextEdit]) -> None: + """ + Apply a list of text edits to a file. + + :param relative_path: The relative path of the file to edit + :param edits: List of TextEdit dictionaries to apply + """ + with self.open_file(relative_path): + # Sort edits by position (latest first) to avoid position shifts + sorted_edits = sorted(edits, key=lambda e: (e["range"]["start"]["line"], e["range"]["start"]["character"]), reverse=True) + + for edit in sorted_edits: + start_pos = ls_types.Position(line=edit["range"]["start"]["line"], character=edit["range"]["start"]["character"]) + end_pos = ls_types.Position(line=edit["range"]["end"]["line"], character=edit["range"]["end"]["character"]) + + # Delete the old text and insert the new text + self.delete_text_between_positions(relative_path, start_pos, end_pos) + self.insert_text_at_position(relative_path, start_pos["line"], start_pos["character"], edit["newText"]) + + def start(self) -> "SolidLanguageServer": + """ + Starts the language server process and connects to it. Call shutdown when ready. + + :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.server_started = True + self._start_server() + return self + + def stop(self, shutdown_timeout: float = 2.0) -> None: + """ + Stops the language server process. + This function never raises an exception (any exceptions during shutdown are logged). + + :param shutdown_timeout: time, in seconds, to wait for the server to shutdown gracefully before killing it + """ + try: + 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) -> LanguageServerInterface: + """Access the underlying language server handler. + + Useful for advanced operations like sending custom commands + or registering notification handlers. + """ + return self.server + + def is_running(self) -> bool: + return self.server.is_running() + + def _create_initialize_params_builder(self) -> InitializeParamsBuilder: + return DefaultInitializeParamsBuilder(self) + + @abstractmethod + def _create_base_initialize_params(self) -> dict | InitializeParams: + """ + Subclasses should override this method to provide server-specific InitializeParams settings, + which provide the basis for the InitializeParams object sent to the language server during initialization. + + The returned dictionary is passed to the builder constructed by _create_initialize_params_builder() + in order to create the final InitializeParams object. + + The default builder implementation already sets the following keys, so implementations of this method + should not set them: + + - processId + - rootPath + - rootUri + - clientInfo + - workspaceFolders + + :return: the base InitializeParams settings + """ + + def _create_initialize_params(self) -> InitializeParams: + """ + Create the InitializeParams object to send to the language server during initialization. + """ + return self._create_initialize_params_builder().with_base_options(self._create_base_initialize_params()).build()