From 58477ffd646cb1a32ef4c4caa150137739e50c78 Mon Sep 17 00:00:00 2001 From: Dominik Jain Date: Tue, 14 Jul 2026 20:11:28 +0200 Subject: [PATCH] Move dependency providers to separate module --- src/solidlsp/dependency_provider.py | 3123 +-------------------------- src/solidlsp/ls.py | 254 +-- 2 files changed, 17 insertions(+), 3360 deletions(-) diff --git a/src/solidlsp/dependency_provider.py b/src/solidlsp/dependency_provider.py index f296fee6..d3826eba 100644 --- a/src/solidlsp/dependency_provider.py +++ b/src/solidlsp/dependency_provider.py @@ -1,276 +1,13 @@ -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 +from collections.abc import Sequence -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): """ @@ -507,2861 +244,3 @@ class LanguageServerDependencyProviderUvx(LanguageServerDependencyProviderBaseCo 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() diff --git a/src/solidlsp/ls.py b/src/solidlsp/ls.py index f296fee6..8731204d 100644 --- a/src/solidlsp/ls.py +++ b/src/solidlsp/ls.py @@ -8,7 +8,7 @@ import shutil import threading from abc import ABC, abstractmethod from collections import defaultdict -from collections.abc import Callable, Hashable, Iterator, Sequence +from collections.abc import Callable, Hashable, Iterator from contextlib import contextmanager from copy import copy from dataclasses import dataclass @@ -17,12 +17,19 @@ from time import monotonic, perf_counter, sleep from typing import Any, Self, Union, cast import pathspec +from sensai.util.helper import mark_used 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.dependency_provider import ( + LanguageServerDependencyProvider, + LanguageServerDependencyProviderBaseCommand, + LanguageServerDependencyProviderSinglePath, + LanguageServerDependencyProviderUvx, +) from solidlsp.initialize_params import DefaultInitializeParamsBuilder, InitializeParamsBuilder from solidlsp.ls_config import FilenameMatcher, Language, LanguageServerConfig from solidlsp.ls_exceptions import SolidLSPException @@ -59,6 +66,14 @@ The `DocumentSymbol` is the preferred type, but the legacy type `SymbolInformati log = logging.getLogger(__name__) +# backward compatibility (support old imports of these classes from this module) +mark_used( + LanguageServerDependencyProvider, + LanguageServerDependencyProviderBaseCommand, + LanguageServerDependencyProviderUvx, + LanguageServerDependencyProviderSinglePath, +) + _debug_enabled = log.isEnabledFor(logging.DEBUG) """Serves as a flag that triggers additional computation when debug logging is enabled.""" @@ -272,243 +287,6 @@ class DocumentSymbols: 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