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 from contextlib import contextmanager from copy import copy from dataclasses import dataclass from pathlib import Path, PurePath from time import monotonic, perf_counter, sleep from typing import Any, Self, Union, cast import pathspec from sensai.util.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 InvalidTextLocationError, 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__) # 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.""" @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: end_line = self._end_line end_col = self._end_col if end_line >= len(self._lines): if end_line == len(self._lines) and end_col == 0: # LSP convention: a range covering whole lines through EOF sometimes ends # at the start of the following, non-existent line (exactly one line past # the last valid index, at column 0). That is well-defined: it means # "through EOF", so treat it as ending at the end of the actual last line. end_line = len(self._lines) - 1 end_col = len(self._lines[end_line]) else: # Any other out-of-range end position (further past EOF, or exactly one # line past EOF but not at column 0) is not the well-defined convention # above; applying the same correction there would silently assume that # a column meant for a nonexistent line still applies to the corrected # one, which can produce a garbage body. Reject it instead of guessing. raise InvalidTextLocationError( f"Symbol range end (line {self._end_line}, col {self._end_col}) is out of bounds " f"for a file with {len(self._lines)} lines" ) # extract relevant lines symbol_body = "\n".join(self._lines[self._start_line : 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[end_line] trailing_length = len(last_line) - end_col if trailing_length > 0: symbol_body = symbol_body[: -(len(last_line) - 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 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 low-level server interface, potentially installing dependencies and launching a subprocess self._process_launch_info: ProcessLaunchInfo | None = process_launch_info self._dependency_provider: LanguageServerDependencyProvider | None = None self.server = self._create_language_server_interface(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 _get_dependency_provider(self) -> LanguageServerDependencyProvider: if self._dependency_provider is None: self._dependency_provider = self._create_dependency_provider() return self._dependency_provider def _create_process_launch_info(self) -> ProcessLaunchInfo: dependency_provider = self._get_dependency_provider() cmd = dependency_provider.create_launch_command() env = dependency_provider.create_launch_command_env() return ProcessLaunchInfo(cmd=cmd, cwd=self.repository_root_path, env=env) def _get_process_launch_info(self) -> ProcessLaunchInfo: if self._process_launch_info is None: self._process_launch_info = self._create_process_launch_info() return self._process_launch_info def _create_language_server_interface(self, 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 logging_fn: the trace logging function :return: the interface """ language_id = self.language_id process_launch_info = self._get_process_launch_info() log.debug(f"Creating language server instance with {language_id=} and {process_launch_info}") 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()