import logging import os import pathlib import stat import subprocess import threading from collections.abc import Hashable from typing import Any from overrides import override from solidlsp.ls import RawDocumentSymbol, SolidLanguageServer from solidlsp.ls_config import Language, LanguageServerConfig from solidlsp.ls_utils import FileUtils, PlatformId, PlatformUtils from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo from solidlsp.settings import SolidLSPSettings from ..common import RuntimeDependency log = logging.getLogger(__name__) EXPERT_VERSION = "v0.1.0-rc.6" EXPERT_ALLOWED_HOSTS = ( "github.com", "release-assets.githubusercontent.com", "objects.githubusercontent.com", ) class ElixirTools(SolidLanguageServer): """ Provides Elixir specific instantiation of the LanguageServer class using Expert, the official Elixir language server. You can pass the following entries in ``ls_specific_settings["elixir"]``: - expert_version: Override the pinned Expert version downloaded by Serena (default: the bundled Serena version). """ @override def _get_wait_time_for_cross_file_referencing(self) -> float: return 10.0 # Elixir projects need time to compile and index before cross-file references work @override def is_ignored_dirname(self, dirname: str) -> bool: # For Elixir projects, we should ignore: # - _build: compiled artifacts # - deps: dependencies # - node_modules: if the project has JavaScript components # - .elixir_ls: ElixirLS artifacts (in case both are present) # - cover: coverage reports # - .expert: Expert artifacts return super().is_ignored_dirname(dirname) or dirname in ["_build", "deps", "node_modules", ".elixir_ls", ".expert", "cover"] @override def is_ignored_path(self, relative_path: str, ignore_unsupported_files: bool = True) -> bool: """Check if a path should be ignored for symbol indexing.""" if relative_path.endswith("mix.exs"): # These are project configuration files, not source code with symbols to index return True return super().is_ignored_path(relative_path, ignore_unsupported_files) @classmethod def _get_elixir_version(cls) -> str | None: """Get the installed Elixir version or None if not found.""" try: result = subprocess.run(["elixir", "--version"], capture_output=True, text=True, check=False) if result.returncode == 0: return result.stdout.strip() except FileNotFoundError: return None return None @classmethod def _setup_runtime_dependencies(cls, config: LanguageServerConfig, solidlsp_settings: SolidLSPSettings) -> str: """ Setup runtime dependencies for Expert. Downloads the Expert binary for the current platform and returns the path to the executable. """ elixir_settings = solidlsp_settings.get_ls_specific_settings(Language.ELIXIR) expert_version = elixir_settings.get("expert_version", EXPERT_VERSION) # Check if Elixir is available first elixir_version = cls._get_elixir_version() if not elixir_version: raise RuntimeError( "Elixir is not installed. Please install Elixir from https://elixir-lang.org/install.html and make sure it is added to your PATH." ) log.info(f"Found Elixir: {elixir_version}") # First, check if expert is already in PATH (user may have installed it manually) import shutil expert_in_path = shutil.which("expert") if expert_in_path: log.info(f"Found Expert in PATH: {expert_in_path}") return expert_in_path platform_id = PlatformUtils.get_platform_id() valid_platforms = [ PlatformId.LINUX_x64, PlatformId.LINUX_arm64, PlatformId.OSX_x64, PlatformId.OSX_arm64, PlatformId.WIN_x64, ] assert platform_id in valid_platforms, f"Platform {platform_id} is not supported for Expert at the moment" expert_dir = os.path.join(cls.ls_resources_dir(solidlsp_settings), "expert") # Define runtime dependencies inline runtime_deps = { PlatformId.LINUX_x64: RuntimeDependency( id="expert_linux_amd64", platform_id="linux-x64", url=f"https://github.com/elixir-lang/expert/releases/download/{expert_version}/expert_linux_amd64", archive_type="binary", binary_name="expert_linux_amd64", extract_path="expert", sha256="643a492ff972246668b0ca356a84c3d0a0f5feeae0ab5dc1b9a126876ed460e4" if expert_version == EXPERT_VERSION else None, allowed_hosts=EXPERT_ALLOWED_HOSTS, ), PlatformId.LINUX_arm64: RuntimeDependency( id="expert_linux_arm64", platform_id="linux-arm64", url=f"https://github.com/elixir-lang/expert/releases/download/{expert_version}/expert_linux_arm64", archive_type="binary", binary_name="expert_linux_arm64", extract_path="expert", sha256="d8b830bdaa8991d7ebf255dacbb3674f3ea335c87d0bfba4b7f907ded4a8f014" if expert_version == EXPERT_VERSION else None, allowed_hosts=EXPERT_ALLOWED_HOSTS, ), PlatformId.OSX_x64: RuntimeDependency( id="expert_darwin_amd64", platform_id="osx-x64", url=f"https://github.com/elixir-lang/expert/releases/download/{expert_version}/expert_darwin_amd64", archive_type="binary", binary_name="expert_darwin_amd64", extract_path="expert", sha256="964f316f1633090b33aab392b6b85fb778c5fb3c0db862671424458da34b1d4d" if expert_version == EXPERT_VERSION else None, allowed_hosts=EXPERT_ALLOWED_HOSTS, ), PlatformId.OSX_arm64: RuntimeDependency( id="expert_darwin_arm64", platform_id="osx-arm64", url=f"https://github.com/elixir-lang/expert/releases/download/{expert_version}/expert_darwin_arm64", archive_type="binary", binary_name="expert_darwin_arm64", extract_path="expert", sha256="5fb5be151baedd635d99835cf3f9986afc9af6ae7b07bd001a1962f4298e45da" if expert_version == EXPERT_VERSION else None, allowed_hosts=EXPERT_ALLOWED_HOSTS, ), PlatformId.WIN_x64: RuntimeDependency( id="expert_windows_amd64", platform_id="win-x64", url=f"https://github.com/elixir-lang/expert/releases/download/{expert_version}/expert_windows_amd64.exe", archive_type="binary", binary_name="expert_windows_amd64.exe", extract_path="expert.exe", sha256="babee77d2653679021600b99c68d984d4463290cb221e0fc0d1093b3afdeb3b0" if expert_version == EXPERT_VERSION else None, allowed_hosts=EXPERT_ALLOWED_HOSTS, ), } dependency = runtime_deps[platform_id] # On Windows, use .exe extension executable_name = "expert.exe" if platform_id.value.startswith("win") else "expert" executable_path = os.path.join(expert_dir, executable_name) assert dependency.binary_name is not None binary_path = os.path.join(expert_dir, dependency.binary_name) if not os.path.exists(executable_path): log.info(f"Downloading Expert binary from {dependency.url}") assert dependency.url is not None FileUtils.download_file_verified( dependency.url, binary_path, expected_sha256=dependency.sha256, allowed_hosts=dependency.allowed_hosts, ) # Make the binary executable on Unix-like systems if not platform_id.value.startswith("win"): os.chmod(binary_path, stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP | stat.S_IROTH | stat.S_IXOTH) # Create a symlink with the expected name on Unix-like systems if binary_path != executable_path and not platform_id.value.startswith("win"): if os.path.exists(executable_path): os.remove(executable_path) os.symlink(os.path.basename(binary_path), executable_path) # normalizing the executable name on Windows if binary_path != executable_path and platform_id.value.startswith("win"): shutil.copy2(binary_path, executable_path) assert os.path.exists(executable_path), f"Expert executable not found at {executable_path}" log.info(f"Expert binary ready at: {executable_path}") return executable_path def __init__(self, config: LanguageServerConfig, repository_root_path: str, solidlsp_settings: SolidLSPSettings): expert_executable_path = self._setup_runtime_dependencies(config, solidlsp_settings) super().__init__( config, repository_root_path, ProcessLaunchInfo(cmd=f"{expert_executable_path} --stdio", cwd=repository_root_path), "elixir", solidlsp_settings, ) self.server_ready = threading.Event() self.request_id = 0 self._building_project = False # Set generous timeout for Expert which can be slow to initialize and respond self.set_request_timeout(180.0) @override def _document_symbols_cache_fingerprint(self) -> Hashable: normalize_symbol_name_version = 1 return normalize_symbol_name_version @override def _normalize_symbol_name(self, symbol: RawDocumentSymbol, relative_file_path: str) -> str: return symbol["name"].removeprefix("defp ").removeprefix("def ").split("(", 1)[0].strip() def _find_mix_exs(self) -> str | None: """ Find mix.exs in the repository. Checks {repository_root_path}/mix.exs first (standard layout), then scans immediate subdirectories for mix.exs (monorepo layout, e.g. server/mix.exs). Returns the absolute path to the first match found, or None if not found. Note: subdirectory iteration order depends on the filesystem (os.listdir). In practice this is fine — monorepos typically have one Elixir app. """ # Check root first root_mix_exs = os.path.join(self.repository_root_path, "mix.exs") if os.path.exists(root_mix_exs): return root_mix_exs # Search immediate subdirectories try: for entry in os.listdir(self.repository_root_path): subdir_path = os.path.join(self.repository_root_path, entry) if os.path.isdir(subdir_path): subdir_mix_exs = os.path.join(subdir_path, "mix.exs") if os.path.exists(subdir_mix_exs): log.info(f"Found mix.exs in subdirectory: {entry}/mix.exs") return subdir_mix_exs except OSError as e: log.warning(f"Error scanning for mix.exs in subdirectories: {e}") return None def _create_base_initialize_params(self) -> dict: """ Returns the initialize params for the Expert Language Server. """ initialize_params = { "locale": "en", "initializationOptions": { "mix_env": "dev", "mix_target": "host", "experimental": {"completions": {"enable": False}}, "extensions": {"credo": {"enable": True, "cli_options": []}}, }, "capabilities": { "textDocument": { "synchronization": {"didSave": True, "dynamicRegistration": True}, "completion": { "dynamicRegistration": True, "completionItem": {"snippetSupport": True, "documentationFormat": ["markdown", "plaintext"]}, }, "definition": {"dynamicRegistration": True}, "references": {"dynamicRegistration": True}, "documentSymbol": { "dynamicRegistration": True, "hierarchicalDocumentSymbolSupport": True, "symbolKind": {"valueSet": list(range(1, 27))}, }, "hover": {"dynamicRegistration": True, "contentFormat": ["markdown", "plaintext"]}, "formatting": {"dynamicRegistration": True}, "codeAction": { "dynamicRegistration": True, "codeActionLiteralSupport": { "codeActionKind": { "valueSet": [ "quickfix", "refactor", "refactor.extract", "refactor.inline", "refactor.rewrite", "source", "source.organizeImports", ] } }, }, }, "workspace": { "workspaceFolders": True, "didChangeConfiguration": {"dynamicRegistration": True}, "executeCommand": {"dynamicRegistration": True}, }, "window": { "showMessage": {"messageActionItem": {"additionalPropertiesSupport": True}}, "showDocument": {"support": True}, "workDoneProgress": True, }, }, } return initialize_params def _start_server(self) -> None: """Start Expert server process""" def register_capability_handler(params: Any) -> None: log.debug(f"LSP: client/registerCapability: {params}") return def window_log_message(msg: Any) -> None: """Handle window/logMessage notifications from Expert""" message_type = msg.get("type", 4) # 1=Error, 2=Warning, 3=Info, 4=Log message_text = msg.get("message", "") # Log at appropriate level based on message type if message_type == 1: log.error(f"Expert: {message_text}") elif message_type == 2: log.warning(f"Expert: {message_text}") else: log.debug(f"Expert: {message_text}") def check_server_ready(params: Any) -> None: """ Handle $/progress notifications from Expert. Expert sends progress updates during compilation and indexing. The server is considered ready when project build completes. """ value = params.get("value", {}) kind = value.get("kind", "") title = value.get("title", "") if kind == "begin": # Track when building the project starts (not "Building engine") if title.startswith("Building ") and not title.startswith("Building engine"): self._building_project = True elif kind == "end": # Project build completion is the main readiness signal if getattr(self, "_building_project", False): log.debug("Expert project build completed - server is ready") self._building_project = False self.server_ready.set() def work_done_progress_create(params: Any) -> None: """Handle window/workDoneProgress/create requests from Expert.""" return def publish_diagnostics(params: Any) -> None: """Handle textDocument/publishDiagnostics notifications.""" return self.server.on_request("client/registerCapability", register_capability_handler) self.server.on_notification("window/logMessage", window_log_message) self.server.on_notification("$/progress", check_server_ready) self.server.on_request("window/workDoneProgress/create", work_done_progress_create) self.server.on_notification("textDocument/publishDiagnostics", publish_diagnostics) log.debug("Starting Expert server process") self.server.start() initialize_params = self._create_initialize_params() log.debug("Sending initialize request to Expert") init_response = self.server.send.initialize(initialize_params) # Verify basic server capabilities assert "textDocumentSync" in init_response["capabilities"], f"Missing textDocumentSync in {init_response['capabilities']}" self.server.notify.initialized({}) # Expert's build pipeline is triggered by a textDocument/didOpen notification. # Without it Expert sits idle and never emits the $/progress signals we wait for, # causing a deadlock. Opening mix.exs is the minimal trigger; we close it again # once the server is ready so it does not linger in the open-file set. mix_exs_path = self._find_mix_exs() mix_exs_uri: str | None = None if mix_exs_path is not None: mix_exs_uri = pathlib.Path(mix_exs_path).as_uri() with open(mix_exs_path, encoding="utf-8") as f: mix_exs_content = f.read() self.server.notify.did_open_text_document( { "textDocument": { "uri": mix_exs_uri, "languageId": "elixir", "version": 1, "text": mix_exs_content, } } ) log.debug("Opened mix.exs to trigger Expert's build pipeline") # Expert needs time to compile the project and build indexes on first run. # This can take 2-3+ minutes for mid-sized codebases. # After the first run, subsequent startups are much faster. ready_timeout = 300.0 # 5 minutes log.debug(f"Waiting up to {ready_timeout}s for Expert to compile and index...") if self.server_ready.wait(timeout=ready_timeout): log.debug("Expert is ready for requests") else: log.warning(f"Expert did not signal readiness within {ready_timeout}s. Proceeding with requests anyway.") self.server_ready.set() # Mark as ready anyway to allow requests if mix_exs_uri is not None: self.server.notify.did_close_text_document({"textDocument": {"uri": mix_exs_uri}}) log.debug("Closed mix.exs after Expert is ready")