This is page 4 of 17. Use http://codebase.md/oraios/serena?lines=false&page={x} to view the full context.
# Directory Structure
```
├── .devcontainer
│ └── devcontainer.json
├── .dockerignore
├── .env.example
├── .github
│ ├── FUNDING.yml
│ ├── ISSUE_TEMPLATE
│ │ ├── config.yml
│ │ ├── feature_request.md
│ │ └── issue--bug--performance-problem--question-.md
│ └── workflows
│ ├── codespell.yml
│ ├── docker.yml
│ ├── docs.yaml
│ ├── junie.yml
│ ├── publish.yml
│ └── pytest.yml
├── .gitignore
├── .serena
│ ├── .gitignore
│ ├── memories
│ │ ├── adding_new_language_support_guide.md
│ │ ├── serena_core_concepts_and_architecture.md
│ │ ├── serena_repository_structure.md
│ │ └── suggested_commands.md
│ └── project.yml
├── .vscode
│ └── settings.json
├── CHANGELOG.md
├── CLAUDE.md
├── compose.yaml
├── CONTRIBUTING.md
├── docker_build_and_run.sh
├── DOCKER.md
├── Dockerfile
├── docs
│ ├── _config.yml
│ ├── _static
│ │ └── images
│ │ └── jetbrains-marketplace-button.png
│ ├── .gitignore
│ ├── 01-about
│ │ ├── 000_intro.md
│ │ ├── 010_llm-integration.md
│ │ ├── 020_programming-languages.md
│ │ ├── 030_serena-in-action.md
│ │ ├── 035_tools.md
│ │ ├── 040_comparison-to-other-agents.md
│ │ └── 050_acknowledgements.md
│ ├── 02-usage
│ │ ├── 000_intro.md
│ │ ├── 010_prerequisites.md
│ │ ├── 020_running.md
│ │ ├── 025_jetbrains_plugin.md
│ │ ├── 030_clients.md
│ │ ├── 040_workflow.md
│ │ ├── 050_configuration.md
│ │ ├── 060_dashboard.md
│ │ ├── 070_security.md
│ │ └── 999_additional-usage.md
│ ├── 03-special-guides
│ │ ├── 000_intro.md
│ │ ├── custom_agent.md
│ │ ├── groovy_setup_guide_for_serena.md
│ │ ├── scala_setup_guide_for_serena.md
│ │ └── serena_on_chatgpt.md
│ ├── autogen_rst.py
│ ├── create_toc.py
│ └── index.md
├── flake.lock
├── flake.nix
├── lessons_learned.md
├── LICENSE
├── llms-install.md
├── pyproject.toml
├── README.md
├── repo_dir_sync.py
├── resources
│ ├── jetbrains-marketplace-button.cdr
│ ├── serena-icons.cdr
│ ├── serena-logo-dark-mode.svg
│ ├── serena-logo.cdr
│ ├── serena-logo.svg
│ └── vscode_sponsor_logo.png
├── roadmap.md
├── scripts
│ ├── agno_agent.py
│ ├── demo_run_tools.py
│ ├── gen_prompt_factory.py
│ ├── mcp_server.py
│ ├── print_mode_context_options.py
│ ├── print_tool_overview.py
│ └── profile_tool_call.py
├── src
│ ├── interprompt
│ │ ├── __init__.py
│ │ ├── .syncCommitId.remote
│ │ ├── .syncCommitId.this
│ │ ├── jinja_template.py
│ │ ├── multilang_prompt.py
│ │ ├── prompt_factory.py
│ │ └── util
│ │ ├── __init__.py
│ │ └── class_decorators.py
│ ├── README.md
│ ├── serena
│ │ ├── __init__.py
│ │ ├── agent.py
│ │ ├── agno.py
│ │ ├── analytics.py
│ │ ├── cli.py
│ │ ├── code_editor.py
│ │ ├── config
│ │ │ ├── __init__.py
│ │ │ ├── context_mode.py
│ │ │ └── serena_config.py
│ │ ├── constants.py
│ │ ├── dashboard.py
│ │ ├── generated
│ │ │ └── generated_prompt_factory.py
│ │ ├── gui_log_viewer.py
│ │ ├── ls_manager.py
│ │ ├── mcp.py
│ │ ├── project.py
│ │ ├── prompt_factory.py
│ │ ├── resources
│ │ │ ├── config
│ │ │ │ ├── contexts
│ │ │ │ │ ├── agent.yml
│ │ │ │ │ ├── chatgpt.yml
│ │ │ │ │ ├── claude-code.yml
│ │ │ │ │ ├── codex.yml
│ │ │ │ │ ├── context.template.yml
│ │ │ │ │ ├── desktop-app.yml
│ │ │ │ │ ├── ide.yml
│ │ │ │ │ └── oaicompat-agent.yml
│ │ │ │ ├── internal_modes
│ │ │ │ │ └── jetbrains.yml
│ │ │ │ ├── modes
│ │ │ │ │ ├── editing.yml
│ │ │ │ │ ├── interactive.yml
│ │ │ │ │ ├── mode.template.yml
│ │ │ │ │ ├── no-memories.yml
│ │ │ │ │ ├── no-onboarding.yml
│ │ │ │ │ ├── onboarding.yml
│ │ │ │ │ ├── one-shot.yml
│ │ │ │ │ └── planning.yml
│ │ │ │ └── prompt_templates
│ │ │ │ ├── simple_tool_outputs.yml
│ │ │ │ └── system_prompt.yml
│ │ │ ├── dashboard
│ │ │ │ ├── dashboard.css
│ │ │ │ ├── dashboard.js
│ │ │ │ ├── index.html
│ │ │ │ ├── jquery.min.js
│ │ │ │ ├── serena-icon-16.png
│ │ │ │ ├── serena-icon-32.png
│ │ │ │ ├── serena-icon-48.png
│ │ │ │ ├── serena-logo-dark-mode.svg
│ │ │ │ ├── serena-logo.svg
│ │ │ │ ├── serena-logs-dark-mode.png
│ │ │ │ └── serena-logs.png
│ │ │ ├── project.template.yml
│ │ │ └── serena_config.template.yml
│ │ ├── symbol.py
│ │ ├── task_executor.py
│ │ ├── text_utils.py
│ │ ├── tools
│ │ │ ├── __init__.py
│ │ │ ├── cmd_tools.py
│ │ │ ├── config_tools.py
│ │ │ ├── file_tools.py
│ │ │ ├── jetbrains_plugin_client.py
│ │ │ ├── jetbrains_tools.py
│ │ │ ├── memory_tools.py
│ │ │ ├── symbol_tools.py
│ │ │ ├── tools_base.py
│ │ │ └── workflow_tools.py
│ │ └── util
│ │ ├── class_decorators.py
│ │ ├── cli_util.py
│ │ ├── exception.py
│ │ ├── file_system.py
│ │ ├── general.py
│ │ ├── git.py
│ │ ├── gui.py
│ │ ├── inspection.py
│ │ ├── logging.py
│ │ ├── shell.py
│ │ └── thread.py
│ └── solidlsp
│ ├── __init__.py
│ ├── .gitignore
│ ├── language_servers
│ │ ├── al_language_server.py
│ │ ├── bash_language_server.py
│ │ ├── clangd_language_server.py
│ │ ├── clojure_lsp.py
│ │ ├── common.py
│ │ ├── csharp_language_server.py
│ │ ├── dart_language_server.py
│ │ ├── eclipse_jdtls.py
│ │ ├── elixir_tools
│ │ │ ├── __init__.py
│ │ │ ├── elixir_tools.py
│ │ │ └── README.md
│ │ ├── elm_language_server.py
│ │ ├── erlang_language_server.py
│ │ ├── fortran_language_server.py
│ │ ├── fsharp_language_server.py
│ │ ├── gopls.py
│ │ ├── groovy_language_server.py
│ │ ├── haskell_language_server.py
│ │ ├── intelephense.py
│ │ ├── jedi_server.py
│ │ ├── julia_server.py
│ │ ├── kotlin_language_server.py
│ │ ├── lua_ls.py
│ │ ├── marksman.py
│ │ ├── matlab_language_server.py
│ │ ├── nixd_ls.py
│ │ ├── omnisharp
│ │ │ ├── initialize_params.json
│ │ │ ├── runtime_dependencies.json
│ │ │ └── workspace_did_change_configuration.json
│ │ ├── omnisharp.py
│ │ ├── pascal_server.py
│ │ ├── perl_language_server.py
│ │ ├── powershell_language_server.py
│ │ ├── pyright_server.py
│ │ ├── r_language_server.py
│ │ ├── regal_server.py
│ │ ├── ruby_lsp.py
│ │ ├── rust_analyzer.py
│ │ ├── scala_language_server.py
│ │ ├── solargraph.py
│ │ ├── sourcekit_lsp.py
│ │ ├── taplo_server.py
│ │ ├── terraform_ls.py
│ │ ├── typescript_language_server.py
│ │ ├── vts_language_server.py
│ │ ├── vue_language_server.py
│ │ ├── yaml_language_server.py
│ │ └── zls.py
│ ├── ls_config.py
│ ├── ls_exceptions.py
│ ├── ls_handler.py
│ ├── ls_request.py
│ ├── ls_types.py
│ ├── ls_utils.py
│ ├── ls.py
│ ├── lsp_protocol_handler
│ │ ├── lsp_constants.py
│ │ ├── lsp_requests.py
│ │ ├── lsp_types.py
│ │ └── server.py
│ ├── settings.py
│ └── util
│ ├── cache.py
│ ├── subprocess_util.py
│ └── zip.py
├── sync.py
├── test
│ ├── __init__.py
│ ├── conftest.py
│ ├── resources
│ │ └── repos
│ │ ├── al
│ │ │ └── test_repo
│ │ │ ├── app.json
│ │ │ └── src
│ │ │ ├── Codeunits
│ │ │ │ ├── CustomerMgt.Codeunit.al
│ │ │ │ └── PaymentProcessorImpl.Codeunit.al
│ │ │ ├── Enums
│ │ │ │ └── CustomerType.Enum.al
│ │ │ ├── Interfaces
│ │ │ │ └── IPaymentProcessor.Interface.al
│ │ │ ├── Pages
│ │ │ │ ├── CustomerCard.Page.al
│ │ │ │ └── CustomerList.Page.al
│ │ │ ├── TableExtensions
│ │ │ │ └── Item.TableExt.al
│ │ │ └── Tables
│ │ │ └── Customer.Table.al
│ │ ├── bash
│ │ │ └── test_repo
│ │ │ ├── config.sh
│ │ │ ├── main.sh
│ │ │ └── utils.sh
│ │ ├── clojure
│ │ │ └── test_repo
│ │ │ ├── deps.edn
│ │ │ └── src
│ │ │ └── test_app
│ │ │ ├── core.clj
│ │ │ └── utils.clj
│ │ ├── csharp
│ │ │ └── test_repo
│ │ │ ├── .gitignore
│ │ │ ├── Models
│ │ │ │ └── Person.cs
│ │ │ ├── Program.cs
│ │ │ ├── serena.sln
│ │ │ └── TestProject.csproj
│ │ ├── dart
│ │ │ └── test_repo
│ │ │ ├── .gitignore
│ │ │ ├── lib
│ │ │ │ ├── helper.dart
│ │ │ │ ├── main.dart
│ │ │ │ └── models.dart
│ │ │ └── pubspec.yaml
│ │ ├── elixir
│ │ │ └── test_repo
│ │ │ ├── .gitignore
│ │ │ ├── lib
│ │ │ │ ├── examples.ex
│ │ │ │ ├── ignored_dir
│ │ │ │ │ └── ignored_module.ex
│ │ │ │ ├── models.ex
│ │ │ │ ├── services.ex
│ │ │ │ ├── test_repo.ex
│ │ │ │ └── utils.ex
│ │ │ ├── mix.exs
│ │ │ ├── mix.lock
│ │ │ ├── scripts
│ │ │ │ └── build_script.ex
│ │ │ └── test
│ │ │ ├── models_test.exs
│ │ │ └── test_repo_test.exs
│ │ ├── elm
│ │ │ └── test_repo
│ │ │ ├── elm.json
│ │ │ ├── Main.elm
│ │ │ └── Utils.elm
│ │ ├── erlang
│ │ │ └── test_repo
│ │ │ ├── hello.erl
│ │ │ ├── ignored_dir
│ │ │ │ └── ignored_module.erl
│ │ │ ├── include
│ │ │ │ ├── records.hrl
│ │ │ │ └── types.hrl
│ │ │ ├── math_utils.erl
│ │ │ ├── rebar.config
│ │ │ ├── src
│ │ │ │ ├── app.erl
│ │ │ │ ├── models.erl
│ │ │ │ ├── services.erl
│ │ │ │ └── utils.erl
│ │ │ └── test
│ │ │ ├── models_tests.erl
│ │ │ └── utils_tests.erl
│ │ ├── fortran
│ │ │ └── test_repo
│ │ │ ├── main.f90
│ │ │ └── modules
│ │ │ ├── geometry.f90
│ │ │ └── math_utils.f90
│ │ ├── fsharp
│ │ │ └── test_repo
│ │ │ ├── .gitignore
│ │ │ ├── Calculator.fs
│ │ │ ├── Models
│ │ │ │ └── Person.fs
│ │ │ ├── Program.fs
│ │ │ ├── README.md
│ │ │ └── TestProject.fsproj
│ │ ├── go
│ │ │ └── test_repo
│ │ │ └── main.go
│ │ ├── groovy
│ │ │ └── test_repo
│ │ │ ├── .gitignore
│ │ │ ├── build.gradle
│ │ │ └── src
│ │ │ └── main
│ │ │ └── groovy
│ │ │ └── com
│ │ │ └── example
│ │ │ ├── Main.groovy
│ │ │ ├── Model.groovy
│ │ │ ├── ModelUser.groovy
│ │ │ └── Utils.groovy
│ │ ├── haskell
│ │ │ └── test_repo
│ │ │ ├── app
│ │ │ │ └── Main.hs
│ │ │ ├── haskell-test-repo.cabal
│ │ │ ├── package.yaml
│ │ │ ├── src
│ │ │ │ ├── Calculator.hs
│ │ │ │ └── Helper.hs
│ │ │ └── stack.yaml
│ │ ├── java
│ │ │ └── test_repo
│ │ │ ├── pom.xml
│ │ │ └── src
│ │ │ └── main
│ │ │ └── java
│ │ │ └── test_repo
│ │ │ ├── Main.java
│ │ │ ├── Model.java
│ │ │ ├── ModelUser.java
│ │ │ └── Utils.java
│ │ ├── julia
│ │ │ └── test_repo
│ │ │ ├── lib
│ │ │ │ └── helper.jl
│ │ │ └── main.jl
│ │ ├── kotlin
│ │ │ └── test_repo
│ │ │ ├── .gitignore
│ │ │ ├── build.gradle.kts
│ │ │ └── src
│ │ │ └── main
│ │ │ └── kotlin
│ │ │ └── test_repo
│ │ │ ├── Main.kt
│ │ │ ├── Model.kt
│ │ │ ├── ModelUser.kt
│ │ │ └── Utils.kt
│ │ ├── lua
│ │ │ └── test_repo
│ │ │ ├── .gitignore
│ │ │ ├── main.lua
│ │ │ ├── src
│ │ │ │ ├── calculator.lua
│ │ │ │ └── utils.lua
│ │ │ └── tests
│ │ │ └── test_calculator.lua
│ │ ├── markdown
│ │ │ └── test_repo
│ │ │ ├── api.md
│ │ │ ├── CONTRIBUTING.md
│ │ │ ├── guide.md
│ │ │ └── README.md
│ │ ├── matlab
│ │ │ └── test_repo
│ │ │ ├── Calculator.m
│ │ │ └── main.m
│ │ ├── nix
│ │ │ └── test_repo
│ │ │ ├── .gitignore
│ │ │ ├── default.nix
│ │ │ ├── flake.nix
│ │ │ ├── lib
│ │ │ │ └── utils.nix
│ │ │ ├── modules
│ │ │ │ └── example.nix
│ │ │ └── scripts
│ │ │ └── hello.sh
│ │ ├── pascal
│ │ │ └── test_repo
│ │ │ ├── .gitignore
│ │ │ ├── lib
│ │ │ │ └── helper.pas
│ │ │ └── main.pas
│ │ ├── perl
│ │ │ └── test_repo
│ │ │ ├── helper.pl
│ │ │ └── main.pl
│ │ ├── php
│ │ │ └── test_repo
│ │ │ ├── helper.php
│ │ │ ├── index.php
│ │ │ └── simple_var.php
│ │ ├── powershell
│ │ │ └── test_repo
│ │ │ ├── main.ps1
│ │ │ ├── PowerShellEditorServices.json
│ │ │ └── utils.ps1
│ │ ├── python
│ │ │ └── test_repo
│ │ │ ├── .gitignore
│ │ │ ├── custom_test
│ │ │ │ ├── __init__.py
│ │ │ │ └── advanced_features.py
│ │ │ ├── examples
│ │ │ │ ├── __init__.py
│ │ │ │ └── user_management.py
│ │ │ ├── ignore_this_dir_with_postfix
│ │ │ │ └── ignored_module.py
│ │ │ ├── scripts
│ │ │ │ ├── __init__.py
│ │ │ │ └── run_app.py
│ │ │ └── test_repo
│ │ │ ├── __init__.py
│ │ │ ├── complex_types.py
│ │ │ ├── models.py
│ │ │ ├── name_collisions.py
│ │ │ ├── nested_base.py
│ │ │ ├── nested.py
│ │ │ ├── overloaded.py
│ │ │ ├── services.py
│ │ │ ├── utils.py
│ │ │ └── variables.py
│ │ ├── r
│ │ │ └── test_repo
│ │ │ ├── .Rbuildignore
│ │ │ ├── DESCRIPTION
│ │ │ ├── examples
│ │ │ │ └── analysis.R
│ │ │ ├── NAMESPACE
│ │ │ └── R
│ │ │ ├── models.R
│ │ │ └── utils.R
│ │ ├── rego
│ │ │ └── test_repo
│ │ │ ├── policies
│ │ │ │ ├── authz.rego
│ │ │ │ └── validation.rego
│ │ │ └── utils
│ │ │ └── helpers.rego
│ │ ├── ruby
│ │ │ └── test_repo
│ │ │ ├── .solargraph.yml
│ │ │ ├── examples
│ │ │ │ └── user_management.rb
│ │ │ ├── lib.rb
│ │ │ ├── main.rb
│ │ │ ├── models.rb
│ │ │ ├── nested.rb
│ │ │ ├── services.rb
│ │ │ └── variables.rb
│ │ ├── rust
│ │ │ ├── test_repo
│ │ │ │ ├── Cargo.lock
│ │ │ │ ├── Cargo.toml
│ │ │ │ └── src
│ │ │ │ ├── lib.rs
│ │ │ │ └── main.rs
│ │ │ └── test_repo_2024
│ │ │ ├── Cargo.lock
│ │ │ ├── Cargo.toml
│ │ │ └── src
│ │ │ ├── lib.rs
│ │ │ └── main.rs
│ │ ├── scala
│ │ │ ├── build.sbt
│ │ │ ├── project
│ │ │ │ ├── build.properties
│ │ │ │ ├── metals.sbt
│ │ │ │ └── plugins.sbt
│ │ │ └── src
│ │ │ └── main
│ │ │ └── scala
│ │ │ └── com
│ │ │ └── example
│ │ │ ├── Main.scala
│ │ │ └── Utils.scala
│ │ ├── swift
│ │ │ └── test_repo
│ │ │ ├── Package.swift
│ │ │ └── src
│ │ │ ├── main.swift
│ │ │ └── utils.swift
│ │ ├── terraform
│ │ │ └── test_repo
│ │ │ ├── data.tf
│ │ │ ├── main.tf
│ │ │ ├── outputs.tf
│ │ │ └── variables.tf
│ │ ├── toml
│ │ │ └── test_repo
│ │ │ ├── Cargo.toml
│ │ │ ├── config.toml
│ │ │ └── pyproject.toml
│ │ ├── typescript
│ │ │ └── test_repo
│ │ │ ├── .serena
│ │ │ │ └── project.yml
│ │ │ ├── index.ts
│ │ │ ├── tsconfig.json
│ │ │ ├── use_helper.ts
│ │ │ └── ws_manager.js
│ │ ├── vue
│ │ │ └── test_repo
│ │ │ ├── .gitignore
│ │ │ ├── index.html
│ │ │ ├── package.json
│ │ │ ├── src
│ │ │ │ ├── App.vue
│ │ │ │ ├── components
│ │ │ │ │ ├── CalculatorButton.vue
│ │ │ │ │ ├── CalculatorDisplay.vue
│ │ │ │ │ └── CalculatorInput.vue
│ │ │ │ ├── composables
│ │ │ │ │ ├── useFormatter.ts
│ │ │ │ │ └── useTheme.ts
│ │ │ │ ├── main.ts
│ │ │ │ ├── stores
│ │ │ │ │ └── calculator.ts
│ │ │ │ └── types
│ │ │ │ └── index.ts
│ │ │ ├── tsconfig.json
│ │ │ ├── tsconfig.node.json
│ │ │ └── vite.config.ts
│ │ ├── yaml
│ │ │ └── test_repo
│ │ │ ├── config.yaml
│ │ │ ├── data.yaml
│ │ │ └── services.yml
│ │ └── zig
│ │ └── test_repo
│ │ ├── .gitignore
│ │ ├── build.zig
│ │ ├── src
│ │ │ ├── calculator.zig
│ │ │ ├── main.zig
│ │ │ └── math_utils.zig
│ │ └── zls.json
│ ├── serena
│ │ ├── __init__.py
│ │ ├── __snapshots__
│ │ │ └── test_symbol_editing.ambr
│ │ ├── config
│ │ │ ├── __init__.py
│ │ │ └── test_serena_config.py
│ │ ├── test_cli_project_commands.py
│ │ ├── test_edit_marker.py
│ │ ├── test_mcp.py
│ │ ├── test_serena_agent.py
│ │ ├── test_symbol_editing.py
│ │ ├── test_symbol.py
│ │ ├── test_task_executor.py
│ │ ├── test_text_utils.py
│ │ ├── test_tool_parameter_types.py
│ │ └── util
│ │ ├── test_exception.py
│ │ └── test_file_system.py
│ └── solidlsp
│ ├── al
│ │ └── test_al_basic.py
│ ├── bash
│ │ ├── __init__.py
│ │ └── test_bash_basic.py
│ ├── clojure
│ │ ├── __init__.py
│ │ └── test_clojure_basic.py
│ ├── csharp
│ │ └── test_csharp_basic.py
│ ├── dart
│ │ ├── __init__.py
│ │ └── test_dart_basic.py
│ ├── elixir
│ │ ├── __init__.py
│ │ ├── conftest.py
│ │ ├── test_elixir_basic.py
│ │ ├── test_elixir_ignored_dirs.py
│ │ ├── test_elixir_integration.py
│ │ └── test_elixir_symbol_retrieval.py
│ ├── elm
│ │ └── test_elm_basic.py
│ ├── erlang
│ │ ├── __init__.py
│ │ ├── conftest.py
│ │ ├── test_erlang_basic.py
│ │ ├── test_erlang_ignored_dirs.py
│ │ └── test_erlang_symbol_retrieval.py
│ ├── fortran
│ │ ├── __init__.py
│ │ └── test_fortran_basic.py
│ ├── fsharp
│ │ └── test_fsharp_basic.py
│ ├── go
│ │ └── test_go_basic.py
│ ├── groovy
│ │ └── test_groovy_basic.py
│ ├── haskell
│ │ ├── __init__.py
│ │ └── test_haskell_basic.py
│ ├── java
│ │ └── test_java_basic.py
│ ├── julia
│ │ └── test_julia_basic.py
│ ├── kotlin
│ │ └── test_kotlin_basic.py
│ ├── lua
│ │ └── test_lua_basic.py
│ ├── markdown
│ │ ├── __init__.py
│ │ └── test_markdown_basic.py
│ ├── matlab
│ │ ├── __init__.py
│ │ └── test_matlab_basic.py
│ ├── nix
│ │ └── test_nix_basic.py
│ ├── pascal
│ │ ├── __init__.py
│ │ └── test_pascal_basic.py
│ ├── perl
│ │ └── test_perl_basic.py
│ ├── php
│ │ └── test_php_basic.py
│ ├── powershell
│ │ ├── __init__.py
│ │ └── test_powershell_basic.py
│ ├── python
│ │ ├── test_python_basic.py
│ │ ├── test_retrieval_with_ignored_dirs.py
│ │ └── test_symbol_retrieval.py
│ ├── r
│ │ ├── __init__.py
│ │ └── test_r_basic.py
│ ├── rego
│ │ └── test_rego_basic.py
│ ├── ruby
│ │ ├── test_ruby_basic.py
│ │ └── test_ruby_symbol_retrieval.py
│ ├── rust
│ │ ├── test_rust_2024_edition.py
│ │ ├── test_rust_analyzer_detection.py
│ │ └── test_rust_basic.py
│ ├── scala
│ │ └── test_scala_language_server.py
│ ├── swift
│ │ └── test_swift_basic.py
│ ├── terraform
│ │ └── test_terraform_basic.py
│ ├── test_lsp_protocol_handler_server.py
│ ├── toml
│ │ ├── __init__.py
│ │ ├── test_toml_basic.py
│ │ ├── test_toml_edge_cases.py
│ │ ├── test_toml_ignored_dirs.py
│ │ └── test_toml_symbol_retrieval.py
│ ├── typescript
│ │ └── test_typescript_basic.py
│ ├── util
│ │ └── test_zip.py
│ ├── vue
│ │ ├── __init__.py
│ │ ├── test_vue_basic.py
│ │ ├── test_vue_error_cases.py
│ │ ├── test_vue_rename.py
│ │ └── test_vue_symbol_retrieval.py
│ ├── yaml_ls
│ │ ├── __init__.py
│ │ └── test_yaml_basic.py
│ └── zig
│ └── test_zig_basic.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/test/solidlsp/php/test_php_basic.py:
--------------------------------------------------------------------------------
```python
from pathlib import Path
import pytest
from solidlsp import SolidLanguageServer
from solidlsp.ls_config import Language
@pytest.mark.php
class TestPhpLanguageServer:
@pytest.mark.parametrize("language_server", [Language.PHP], indirect=True)
@pytest.mark.parametrize("repo_path", [Language.PHP], indirect=True)
def test_ls_is_running(self, language_server: SolidLanguageServer, repo_path: Path) -> None:
"""Test that the language server starts and stops successfully."""
# The fixture already handles start and stop
assert language_server.is_running()
assert Path(language_server.language_server.repository_root_path).resolve() == repo_path.resolve()
@pytest.mark.parametrize("language_server", [Language.PHP], indirect=True)
@pytest.mark.parametrize("repo_path", [Language.PHP], indirect=True)
def test_find_definition_within_file(self, language_server: SolidLanguageServer, repo_path: Path) -> None:
# In index.php:
# Line 9 (1-indexed): $greeting = greet($userName);
# Line 11 (1-indexed): echo $greeting;
# We want to find the definition of $greeting (defined on line 9)
# from its usage in echo $greeting; on line 11.
# LSP is 0-indexed: definition on line 8, usage on line 10.
# $greeting in echo $greeting; is at char 5 on line 11 (0-indexed: line 10, char 5)
# e c h o $ g r e e t i n g
# ^ char 5
definition_location_list = language_server.request_definition(str(repo_path / "index.php"), 10, 6) # cursor on 'g' in $greeting
assert definition_location_list, f"Expected non-empty definition_location_list but got {definition_location_list=}"
assert len(definition_location_list) == 1
definition_location = definition_location_list[0]
assert definition_location["uri"].endswith("index.php")
# Definition of $greeting is on line 10 (1-indexed) / line 9 (0-indexed), char 0
assert definition_location["range"]["start"]["line"] == 9
assert definition_location["range"]["start"]["character"] == 0
@pytest.mark.parametrize("language_server", [Language.PHP], indirect=True)
@pytest.mark.parametrize("repo_path", [Language.PHP], indirect=True)
def test_find_definition_across_files(self, language_server: SolidLanguageServer, repo_path: Path) -> None:
definition_location_list = language_server.request_definition(str(repo_path / "index.php"), 12, 5) # helperFunction
assert definition_location_list, f"Expected non-empty definition_location_list but got {definition_location_list=}"
assert len(definition_location_list) == 1
definition_location = definition_location_list[0]
assert definition_location["uri"].endswith("helper.php")
assert definition_location["range"]["start"]["line"] == 2
assert definition_location["range"]["start"]["character"] == 0
@pytest.mark.parametrize("language_server", [Language.PHP], indirect=True)
@pytest.mark.parametrize("repo_path", [Language.PHP], indirect=True)
def test_find_definition_simple_variable(self, language_server: SolidLanguageServer, repo_path: Path) -> None:
file_path = str(repo_path / "simple_var.php")
# In simple_var.php:
# Line 2 (1-indexed): $localVar = "test";
# Line 3 (1-indexed): echo $localVar;
# LSP is 0-indexed: definition on line 1, usage on line 2
# Find definition of $localVar (char 5 on line 3 / 0-indexed: line 2, char 5)
# $localVar in echo $localVar; (e c h o $ l o c a l V a r)
# ^ char 5
definition_location_list = language_server.request_definition(file_path, 2, 6) # cursor on 'l' in $localVar
assert definition_location_list, f"Expected non-empty definition_location_list but got {definition_location_list=}"
assert len(definition_location_list) == 1
definition_location = definition_location_list[0]
assert definition_location["uri"].endswith("simple_var.php")
assert definition_location["range"]["start"]["line"] == 1 # Definition of $localVar (0-indexed)
assert definition_location["range"]["start"]["character"] == 0 # $localVar (0-indexed)
@pytest.mark.parametrize("language_server", [Language.PHP], indirect=True)
@pytest.mark.parametrize("repo_path", [Language.PHP], indirect=True)
def test_find_references_within_file(self, language_server: SolidLanguageServer, repo_path: Path) -> None:
index_php_path = str(repo_path / "index.php")
# In index.php (0-indexed lines):
# Line 9: $greeting = greet($userName); // Definition of $greeting
# Line 11: echo $greeting; // Usage of $greeting
# Find references for $greeting from its usage in "echo $greeting;" (line 11, char 6 for 'g')
references = language_server.request_references(index_php_path, 11, 6)
assert references
# Intelephense, when asked for references from usage, seems to only return the usage itself.
assert len(references) == 1, "Expected to find 1 reference for $greeting (the usage itself)"
expected_locations = [{"uri_suffix": "index.php", "line": 11, "character": 5}] # Usage: echo $greeting (points to $)
# Convert actual references to a comparable format and sort
actual_locations = sorted(
[
{
"uri_suffix": loc["uri"].split("/")[-1],
"line": loc["range"]["start"]["line"],
"character": loc["range"]["start"]["character"],
}
for loc in references
],
key=lambda x: (x["uri_suffix"], x["line"], x["character"]),
)
expected_locations = sorted(expected_locations, key=lambda x: (x["uri_suffix"], x["line"], x["character"]))
assert actual_locations == expected_locations
@pytest.mark.parametrize("language_server", [Language.PHP], indirect=True)
@pytest.mark.parametrize("repo_path", [Language.PHP], indirect=True)
def test_find_references_across_files(self, language_server: SolidLanguageServer, repo_path: Path) -> None:
helper_php_path = str(repo_path / "helper.php")
# In index.php (0-indexed lines):
# Line 13: helperFunction(); // Usage of helperFunction
# Find references for helperFunction from its definition
references = language_server.request_references(helper_php_path, 2, len("function "))
assert references, f"Expected non-empty references for helperFunction but got {references=}"
# Intelephense might return 1 (usage) or 2 (usage + definition) references.
# Let's check for at least the usage in index.php
# Definition is in helper.php, line 2, char 0 (based on previous findings)
# Usage is in index.php, line 13, char 0
actual_locations_comparable = []
for loc in references:
actual_locations_comparable.append(
{
"uri_suffix": loc["uri"].split("/")[-1],
"line": loc["range"]["start"]["line"],
"character": loc["range"]["start"]["character"],
}
)
usage_in_index_php = {"uri_suffix": "index.php", "line": 13, "character": 0}
assert usage_in_index_php in actual_locations_comparable, "Usage of helperFunction in index.php not found"
```
--------------------------------------------------------------------------------
/src/solidlsp/language_servers/marksman.py:
--------------------------------------------------------------------------------
```python
"""
Provides Markdown specific instantiation of the LanguageServer class using marksman.
Contains various configurations and settings specific to Markdown.
"""
import logging
import os
import pathlib
import threading
from overrides import override
from solidlsp.ls import SolidLanguageServer
from solidlsp.ls_config import LanguageServerConfig
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.settings import SolidLSPSettings
from .common import RuntimeDependency, RuntimeDependencyCollection
log = logging.getLogger(__name__)
class Marksman(SolidLanguageServer):
"""
Provides Markdown specific instantiation of the LanguageServer class using marksman.
"""
marksman_releases = "https://github.com/artempyanykh/marksman/releases/download/2024-12-18"
runtime_dependencies = RuntimeDependencyCollection(
[
RuntimeDependency(
id="marksman",
url=f"{marksman_releases}/marksman-linux-x64",
platform_id="linux-x64",
archive_type="binary",
binary_name="marksman",
),
RuntimeDependency(
id="marksman",
url=f"{marksman_releases}/marksman-linux-arm64",
platform_id="linux-arm64",
archive_type="binary",
binary_name="marksman",
),
RuntimeDependency(
id="marksman",
url=f"{marksman_releases}/marksman-macos",
platform_id="osx-x64",
archive_type="binary",
binary_name="marksman",
),
RuntimeDependency(
id="marksman",
url=f"{marksman_releases}/marksman-macos",
platform_id="osx-arm64",
archive_type="binary",
binary_name="marksman",
),
RuntimeDependency(
id="marksman",
url=f"{marksman_releases}/marksman.exe",
platform_id="win-x64",
archive_type="binary",
binary_name="marksman.exe",
),
]
)
@classmethod
def _setup_runtime_dependencies(cls, config: LanguageServerConfig, solidlsp_settings: SolidLSPSettings) -> str:
"""Setup runtime dependencies for marksman and return the command to start the server."""
deps = cls.runtime_dependencies
dependency = deps.get_single_dep_for_current_platform()
marksman_ls_dir = cls.ls_resources_dir(solidlsp_settings)
marksman_executable_path = deps.binary_path(marksman_ls_dir)
if not os.path.exists(marksman_executable_path):
log.info(
f"Downloading marksman from {dependency.url} to {marksman_ls_dir}",
)
deps.install(marksman_ls_dir)
if not os.path.exists(marksman_executable_path):
raise FileNotFoundError(f"Download failed? Could not find marksman executable at {marksman_executable_path}")
os.chmod(marksman_executable_path, 0o755)
return marksman_executable_path
def __init__(self, config: LanguageServerConfig, repository_root_path: str, solidlsp_settings: SolidLSPSettings):
"""
Creates a Marksman instance. This class is not meant to be instantiated directly.
Use LanguageServer.create() instead.
"""
marksman_executable_path = self._setup_runtime_dependencies(config, solidlsp_settings)
super().__init__(
config,
repository_root_path,
ProcessLaunchInfo(cmd=f"{marksman_executable_path} server", cwd=repository_root_path),
"markdown",
solidlsp_settings,
)
self.server_ready = threading.Event()
@override
def is_ignored_dirname(self, dirname: str) -> bool:
return super().is_ignored_dirname(dirname) or dirname in ["node_modules", ".obsidian", ".vitepress", ".vuepress"]
@staticmethod
def _get_initialize_params(repository_absolute_path: str) -> InitializeParams:
"""
Returns the initialize params for the Marksman Language Server.
"""
root_uri = pathlib.Path(repository_absolute_path).as_uri()
initialize_params: InitializeParams = { # type: ignore
"processId": os.getpid(),
"locale": "en",
"rootPath": repository_absolute_path,
"rootUri": root_uri,
"capabilities": {
"textDocument": {
"synchronization": {"didSave": True, "dynamicRegistration": True},
"completion": {"dynamicRegistration": True, "completionItem": {"snippetSupport": True}},
"definition": {"dynamicRegistration": True},
"references": {"dynamicRegistration": True},
"documentSymbol": {
"dynamicRegistration": True,
"hierarchicalDocumentSymbolSupport": True,
"symbolKind": {"valueSet": list(range(1, 27))}, # type: ignore[arg-type]
},
"hover": {"dynamicRegistration": True, "contentFormat": ["markdown", "plaintext"]}, # type: ignore[list-item]
"codeAction": {"dynamicRegistration": True},
},
"workspace": {
"workspaceFolders": True,
"didChangeConfiguration": {"dynamicRegistration": True},
"symbol": {"dynamicRegistration": True},
},
},
"workspaceFolders": [
{
"uri": root_uri,
"name": os.path.basename(repository_absolute_path),
}
],
}
return initialize_params
def _start_server(self) -> None:
"""
Starts the Marksman Language Server and waits for it to be ready.
"""
def register_capability_handler(_params: dict) -> None:
return
def window_log_message(msg: dict) -> None:
log.info(f"LSP: window/logMessage: {msg}")
def do_nothing(_params: dict) -> None:
return
self.server.on_request("client/registerCapability", register_capability_handler)
self.server.on_notification("window/logMessage", window_log_message)
self.server.on_notification("$/progress", do_nothing)
self.server.on_notification("textDocument/publishDiagnostics", do_nothing)
log.info("Starting marksman server process")
self.server.start()
initialize_params = self._get_initialize_params(self.repository_root_path)
log.info("Sending initialize request from LSP client to marksman server and awaiting response")
init_response = self.server.send.initialize(initialize_params)
log.debug(f"Received initialize response from marksman server: {init_response}")
# Verify server capabilities
assert "textDocumentSync" in init_response["capabilities"]
assert "completionProvider" in init_response["capabilities"]
assert "definitionProvider" in init_response["capabilities"]
self.server.notify.initialized({})
# marksman is typically ready immediately after initialization
log.info("Marksman server initialization complete")
self.server_ready.set()
self.completions_available.set()
```
--------------------------------------------------------------------------------
/test/solidlsp/elixir/test_elixir_integration.py:
--------------------------------------------------------------------------------
```python
"""
Integration tests for Elixir language server with test repository.
These tests verify that the language server works correctly with a real Elixir project
and can perform advanced operations like cross-file symbol resolution.
"""
import os
from pathlib import Path
import pytest
from serena.project import Project
from solidlsp import SolidLanguageServer
from solidlsp.ls_config import Language
from . import EXPERT_UNAVAILABLE, EXPERT_UNAVAILABLE_REASON
# These marks will be applied to all tests in this module
pytestmark = [pytest.mark.elixir, pytest.mark.skipif(EXPERT_UNAVAILABLE, reason=f"Next LS not available: {EXPERT_UNAVAILABLE_REASON}")]
class TestElixirIntegration:
"""Integration tests for Elixir language server with test repository."""
@pytest.fixture
def elixir_test_repo_path(self):
"""Get the path to the Elixir test repository."""
test_dir = Path(__file__).parent.parent.parent
return str(test_dir / "resources" / "repos" / "elixir" / "test_repo")
def test_elixir_repo_structure(self, elixir_test_repo_path):
"""Test that the Elixir test repository has the expected structure."""
repo_path = Path(elixir_test_repo_path)
# Check that key files exist
assert (repo_path / "mix.exs").exists(), "mix.exs should exist"
assert (repo_path / "lib" / "test_repo.ex").exists(), "main module should exist"
assert (repo_path / "lib" / "utils.ex").exists(), "utils module should exist"
assert (repo_path / "lib" / "models.ex").exists(), "models module should exist"
assert (repo_path / "lib" / "services.ex").exists(), "services module should exist"
assert (repo_path / "lib" / "examples.ex").exists(), "examples module should exist"
assert (repo_path / "test" / "test_repo_test.exs").exists(), "test file should exist"
assert (repo_path / "test" / "models_test.exs").exists(), "models test should exist"
@pytest.mark.parametrize("language_server", [Language.ELIXIR], indirect=True)
def test_cross_file_symbol_resolution(self, language_server: SolidLanguageServer):
"""Test that symbols can be resolved across different files."""
# Test that User struct from models.ex can be found when referenced in services.ex
services_file = os.path.join("lib", "services.ex")
# Find where User is referenced in services.ex
content = language_server.retrieve_full_file_content(services_file)
lines = content.split("\n")
user_reference_line = None
for i, line in enumerate(lines):
if "alias TestRepo.Models.{User" in line:
user_reference_line = i
break
if user_reference_line is None:
pytest.skip("Could not find User reference in services.ex")
# Try to find the definition
defining_symbol = language_server.request_defining_symbol(services_file, user_reference_line, 30)
if defining_symbol and "location" in defining_symbol:
# Should point to models.ex
assert "models.ex" in defining_symbol["location"]["uri"]
@pytest.mark.parametrize("language_server", [Language.ELIXIR], indirect=True)
def test_module_hierarchy_understanding(self, language_server: SolidLanguageServer):
"""Test that the language server understands Elixir module hierarchy."""
models_file = os.path.join("lib", "models.ex")
symbols = language_server.request_document_symbols(models_file).get_all_symbols_and_roots()
if symbols:
# Flatten symbol structure
all_symbols = []
for symbol_group in symbols:
if isinstance(symbol_group, list):
all_symbols.extend(symbol_group)
else:
all_symbols.append(symbol_group)
symbol_names = [s.get("name", "") for s in all_symbols]
# Should understand nested module structure
expected_modules = ["TestRepo.Models", "User", "Item", "Order"]
found_modules = [name for name in expected_modules if any(name in symbol_name for symbol_name in symbol_names)]
assert len(found_modules) > 0, f"Expected modules {expected_modules}, found symbols {symbol_names}"
def test_file_extension_matching(self):
"""Test that the Elixir language recognizes the correct file extensions."""
language = Language.ELIXIR
matcher = language.get_source_fn_matcher()
# Test Elixir file extensions
assert matcher.is_relevant_filename("lib/test_repo.ex")
assert matcher.is_relevant_filename("test/test_repo_test.exs")
assert matcher.is_relevant_filename("config/config.exs")
assert matcher.is_relevant_filename("mix.exs")
assert matcher.is_relevant_filename("lib/models.ex")
assert matcher.is_relevant_filename("lib/services.ex")
# Test non-Elixir files
assert not matcher.is_relevant_filename("README.md")
assert not matcher.is_relevant_filename("lib/test_repo.py")
assert not matcher.is_relevant_filename("package.json")
assert not matcher.is_relevant_filename("Cargo.toml")
class TestElixirProject:
@pytest.mark.parametrize("project", [Language.ELIXIR], indirect=True)
def test_comprehensive_symbol_search(self, project: Project):
"""Test comprehensive symbol search across the entire project."""
# Search for all function definitions
function_pattern = r"def\s+\w+\s*[\(\s]"
function_matches = project.search_source_files_for_pattern(function_pattern)
# Should find functions across multiple files
if function_matches:
files_with_functions = set()
for match in function_matches:
if match.source_file_path:
files_with_functions.add(os.path.basename(match.source_file_path))
# Should find functions in multiple files
expected_files = {"models.ex", "services.ex", "examples.ex", "utils.ex", "test_repo.ex"}
found_files = expected_files.intersection(files_with_functions)
assert len(found_files) > 0, f"Expected functions in {expected_files}, found in {files_with_functions}"
# Search for struct definitions
struct_pattern = r"defstruct\s+\["
struct_matches = project.search_source_files_for_pattern(struct_pattern)
if struct_matches:
# Should find structs primarily in models.ex
models_structs = [m for m in struct_matches if m.source_file_path and "models.ex" in m.source_file_path]
assert len(models_structs) > 0, "Should find struct definitions in models.ex"
@pytest.mark.parametrize("project", [Language.ELIXIR], indirect=True)
def test_protocol_and_implementation_understanding(self, project: Project):
"""Test that the language server understands Elixir protocols and implementations."""
# Search for protocol definitions
protocol_pattern = r"defprotocol\s+\w+"
protocol_matches = project.search_source_files_for_pattern(protocol_pattern, paths_include_glob="**/models.ex")
if protocol_matches:
# Should find the Serializable protocol
serializable_matches = [m for m in protocol_matches if "Serializable" in str(m)]
assert len(serializable_matches) > 0, "Should find Serializable protocol definition"
# Search for protocol implementations
impl_pattern = r"defimpl\s+\w+"
impl_matches = project.search_source_files_for_pattern(impl_pattern, paths_include_glob="**/models.ex")
if impl_matches:
# Should find multiple implementations
assert len(impl_matches) >= 3, f"Should find at least 3 protocol implementations, found {len(impl_matches)}"
```
--------------------------------------------------------------------------------
/test/solidlsp/elixir/test_elixir_ignored_dirs.py:
--------------------------------------------------------------------------------
```python
import os
from collections.abc import Generator
from pathlib import Path
import pytest
from solidlsp import SolidLanguageServer
from solidlsp.ls_config import Language
from test.conftest import start_ls_context
from . import EXPERT_UNAVAILABLE, EXPERT_UNAVAILABLE_REASON
# These marks will be applied to all tests in this module
pytestmark = [pytest.mark.elixir, pytest.mark.skipif(EXPERT_UNAVAILABLE, reason=f"Expert not available: {EXPERT_UNAVAILABLE_REASON}")]
# Skip slow tests in CI - they require multiple Expert instances which is too slow
IN_CI = bool(os.environ.get("CI") or os.environ.get("GITHUB_ACTIONS"))
SKIP_SLOW_IN_CI = pytest.mark.skipif(
IN_CI,
reason="Slow tests skipped in CI - require multiple Expert instances (~60-90s each)",
)
@pytest.fixture(scope="session")
def ls_with_ignored_dirs() -> Generator[SolidLanguageServer, None, None]:
"""Fixture to set up an LS for the elixir test repo with the 'scripts' directory ignored.
Uses session scope to avoid restarting Expert for each test.
"""
ignored_paths = ["scripts", "ignored_dir"]
with start_ls_context(language=Language.ELIXIR, ignored_paths=ignored_paths) as ls:
yield ls
@pytest.mark.slow
@SKIP_SLOW_IN_CI
def test_symbol_tree_ignores_dir(ls_with_ignored_dirs: SolidLanguageServer):
"""Tests that request_full_symbol_tree ignores the configured directory.
Note: This test uses a separate Expert instance with custom ignored paths,
which adds ~60-90s startup time.
"""
root = ls_with_ignored_dirs.request_full_symbol_tree()[0]
root_children = root["children"]
children_names = {child["name"] for child in root_children}
# Should have lib and test directories, but not scripts or ignored_dir
expected_dirs = {"lib", "test"}
assert expected_dirs.issubset(children_names), f"Expected {expected_dirs} to be in {children_names}"
assert "scripts" not in children_names, f"scripts should not be in {children_names}"
assert "ignored_dir" not in children_names, f"ignored_dir should not be in {children_names}"
@pytest.mark.slow
@SKIP_SLOW_IN_CI
def test_find_references_ignores_dir(ls_with_ignored_dirs: SolidLanguageServer):
"""Tests that find_references ignores the configured directory.
Note: This test uses a separate Expert instance with custom ignored paths,
which adds ~60-90s startup time.
"""
# Location of User struct, which is referenced in scripts and ignored_dir
definition_file = "lib/models.ex"
# Find the User struct definition
symbols = ls_with_ignored_dirs.request_document_symbols(definition_file).get_all_symbols_and_roots()
user_symbol = None
for symbol_group in symbols:
user_symbol = next((s for s in symbol_group if "User" in s.get("name", "")), None)
if user_symbol:
break
if not user_symbol or "selectionRange" not in user_symbol:
pytest.skip("User symbol not found for reference testing")
sel_start = user_symbol["selectionRange"]["start"]
references = ls_with_ignored_dirs.request_references(definition_file, sel_start["line"], sel_start["character"])
# Assert that scripts and ignored_dir do not appear in the references
assert not any("scripts" in ref["relativePath"] for ref in references), "scripts should be ignored"
assert not any("ignored_dir" in ref["relativePath"] for ref in references), "ignored_dir should be ignored"
@pytest.mark.slow
@SKIP_SLOW_IN_CI
@pytest.mark.parametrize("repo_path", [Language.ELIXIR], indirect=True)
def test_refs_and_symbols_with_glob_patterns(repo_path: Path) -> None:
"""Tests that refs and symbols with glob patterns are ignored.
Note: This test uses a separate Expert instance with custom ignored paths,
which adds ~60-90s startup time.
"""
ignored_paths = ["*cripts", "ignored_*"] # codespell:ignore cripts
with start_ls_context(language=Language.ELIXIR, repo_path=str(repo_path), ignored_paths=ignored_paths) as ls:
# Same as in the above tests
root = ls.request_full_symbol_tree()[0]
root_children = root["children"]
children_names = {child["name"] for child in root_children}
# Should have lib and test directories, but not scripts or ignored_dir
expected_dirs = {"lib", "test"}
assert expected_dirs.issubset(children_names), f"Expected {expected_dirs} to be in {children_names}"
assert "scripts" not in children_names, f"scripts should not be in {children_names} (glob pattern)"
assert "ignored_dir" not in children_names, f"ignored_dir should not be in {children_names} (glob pattern)"
# Test that the refs and symbols with glob patterns are ignored
definition_file = "lib/models.ex"
# Find the User struct definition
symbols = ls.request_document_symbols(definition_file).get_all_symbols_and_roots()
user_symbol = None
for symbol_group in symbols:
user_symbol = next((s for s in symbol_group if "User" in s.get("name", "")), None)
if user_symbol:
break
if user_symbol and "selectionRange" in user_symbol:
sel_start = user_symbol["selectionRange"]["start"]
references = ls.request_references(definition_file, sel_start["line"], sel_start["character"])
# Assert that scripts and ignored_dir do not appear in references
assert not any("scripts" in ref["relativePath"] for ref in references), "scripts should be ignored (glob)"
assert not any("ignored_dir" in ref["relativePath"] for ref in references), "ignored_dir should be ignored (glob)"
@pytest.mark.parametrize("language_server", [Language.ELIXIR], indirect=True)
def test_default_ignored_directories(language_server: SolidLanguageServer):
"""Test that default Elixir directories are ignored."""
# Test that Elixir-specific directories are ignored by default
assert language_server.is_ignored_dirname("_build"), "_build should be ignored"
assert language_server.is_ignored_dirname("deps"), "deps should be ignored"
assert language_server.is_ignored_dirname(".elixir_ls"), ".elixir_ls should be ignored"
assert language_server.is_ignored_dirname("cover"), "cover should be ignored"
assert language_server.is_ignored_dirname("node_modules"), "node_modules should be ignored"
# Test that important directories are not ignored
assert not language_server.is_ignored_dirname("lib"), "lib should not be ignored"
assert not language_server.is_ignored_dirname("test"), "test should not be ignored"
assert not language_server.is_ignored_dirname("config"), "config should not be ignored"
assert not language_server.is_ignored_dirname("priv"), "priv should not be ignored"
@pytest.mark.xfail(
reason="Expert 0.1.0 bug: document_symbols may return nil for some files (flaky)",
raises=Exception,
)
@pytest.mark.parametrize("language_server", [Language.ELIXIR], indirect=True)
def test_symbol_tree_excludes_build_dirs(language_server: SolidLanguageServer):
"""Test that symbol tree excludes build and dependency directories."""
symbol_tree = language_server.request_full_symbol_tree()
if symbol_tree:
root = symbol_tree[0]
children_names = {child["name"] for child in root.get("children", [])}
# Build and dependency directories should not appear
ignored_dirs = {"_build", "deps", ".elixir_ls", "cover", "node_modules"}
found_ignored = ignored_dirs.intersection(children_names)
assert len(found_ignored) == 0, f"Found ignored directories in symbol tree: {found_ignored}"
# Important directories should appear
important_dirs = {"lib", "test"}
found_important = important_dirs.intersection(children_names)
assert len(found_important) > 0, f"Expected to find important directories: {important_dirs}, got: {children_names}"
```
--------------------------------------------------------------------------------
/src/solidlsp/language_servers/yaml_language_server.py:
--------------------------------------------------------------------------------
```python
"""
Provides YAML specific instantiation of the LanguageServer class using yaml-language-server.
Contains various configurations and settings specific to YAML files.
"""
import logging
import os
import pathlib
import shutil
import threading
from typing import Any
from solidlsp.language_servers.common import RuntimeDependency, RuntimeDependencyCollection
from solidlsp.ls import SolidLanguageServer
from solidlsp.ls_config import LanguageServerConfig
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.settings import SolidLSPSettings
log = logging.getLogger(__name__)
class YamlLanguageServer(SolidLanguageServer):
"""
Provides YAML specific instantiation of the LanguageServer class using yaml-language-server.
Contains various configurations and settings specific to YAML files.
"""
@staticmethod
def _determine_log_level(line: str) -> int:
"""Classify yaml-language-server stderr output to avoid false-positive errors."""
line_lower = line.lower()
# Known informational messages from yaml-language-server that aren't critical errors
if any(
[
"cannot find module" in line_lower and "package.json" in line_lower, # Schema resolution - not critical
"no parser" in line_lower, # Parser messages - informational
]
):
return logging.DEBUG
return SolidLanguageServer._determine_log_level(line)
def __init__(self, config: LanguageServerConfig, repository_root_path: str, solidlsp_settings: SolidLSPSettings):
"""
Creates a YamlLanguageServer instance. This class is not meant to be instantiated directly.
Use LanguageServer.create() instead.
"""
yaml_lsp_executable_path = self._setup_runtime_dependencies(config, solidlsp_settings)
super().__init__(
config,
repository_root_path,
ProcessLaunchInfo(cmd=yaml_lsp_executable_path, cwd=repository_root_path),
"yaml",
solidlsp_settings,
)
self.server_ready = threading.Event()
@classmethod
def _setup_runtime_dependencies(cls, config: LanguageServerConfig, solidlsp_settings: SolidLSPSettings) -> str:
"""
Setup runtime dependencies for YAML Language Server and return the command to start the server.
"""
# Verify both node and npm are installed
is_node_installed = shutil.which("node") is not None
assert is_node_installed, "node is not installed or isn't in PATH. Please install NodeJS and try again."
is_npm_installed = shutil.which("npm") is not None
assert is_npm_installed, "npm is not installed or isn't in PATH. Please install npm and try again."
deps = RuntimeDependencyCollection(
[
RuntimeDependency(
id="yaml-language-server",
description="yaml-language-server package (Red Hat)",
command="npm install --prefix ./ [email protected]",
platform_id="any",
),
]
)
# Install yaml-language-server if not already installed
yaml_ls_dir = os.path.join(cls.ls_resources_dir(solidlsp_settings), "yaml-lsp")
yaml_executable_path = os.path.join(yaml_ls_dir, "node_modules", ".bin", "yaml-language-server")
# Handle Windows executable extension
if os.name == "nt":
yaml_executable_path += ".cmd"
if not os.path.exists(yaml_executable_path):
log.info(f"YAML Language Server executable not found at {yaml_executable_path}. Installing...")
deps.install(yaml_ls_dir)
log.info("YAML language server dependencies installed successfully")
if not os.path.exists(yaml_executable_path):
raise FileNotFoundError(
f"yaml-language-server executable not found at {yaml_executable_path}, something went wrong with the installation."
)
return f"{yaml_executable_path} --stdio"
@staticmethod
def _get_initialize_params(repository_absolute_path: str) -> InitializeParams:
"""
Returns the initialize params for the YAML Language Server.
"""
root_uri = pathlib.Path(repository_absolute_path).as_uri()
initialize_params = {
"locale": "en",
"capabilities": {
"textDocument": {
"synchronization": {"didSave": True, "dynamicRegistration": True},
"completion": {"dynamicRegistration": True, "completionItem": {"snippetSupport": True}},
"definition": {"dynamicRegistration": True},
"references": {"dynamicRegistration": True},
"documentSymbol": {
"dynamicRegistration": True,
"hierarchicalDocumentSymbolSupport": True,
"symbolKind": {"valueSet": list(range(1, 27))},
},
"hover": {"dynamicRegistration": True, "contentFormat": ["markdown", "plaintext"]},
"codeAction": {"dynamicRegistration": True},
},
"workspace": {
"workspaceFolders": True,
"didChangeConfiguration": {"dynamicRegistration": True},
"symbol": {"dynamicRegistration": True},
},
},
"processId": os.getpid(),
"rootPath": repository_absolute_path,
"rootUri": root_uri,
"workspaceFolders": [
{
"uri": root_uri,
"name": os.path.basename(repository_absolute_path),
}
],
"initializationOptions": {
"yaml": {
"schemaStore": {"enable": True, "url": "https://www.schemastore.org/api/json/catalog.json"},
"format": {"enable": True},
"validate": True,
"hover": True,
"completion": True,
}
},
}
return initialize_params # type: ignore
def _start_server(self) -> None:
"""
Starts the YAML Language Server, waits for the server to be ready and yields the LanguageServer instance.
"""
def register_capability_handler(params: Any) -> None:
return
def do_nothing(params: Any) -> None:
return
def window_log_message(msg: dict) -> None:
log.info(f"LSP: window/logMessage: {msg}")
self.server.on_request("client/registerCapability", register_capability_handler)
self.server.on_notification("window/logMessage", window_log_message)
self.server.on_notification("$/progress", do_nothing)
self.server.on_notification("textDocument/publishDiagnostics", do_nothing)
log.info("Starting YAML server process")
self.server.start()
initialize_params = self._get_initialize_params(self.repository_root_path)
log.info("Sending initialize request from LSP client to LSP server and awaiting response")
init_response = self.server.send.initialize(initialize_params)
log.debug(f"Received initialize response from YAML server: {init_response}")
# Verify document symbol support is available
if "documentSymbolProvider" in init_response["capabilities"]:
log.info("YAML server supports document symbols")
else:
log.warning("Warning: YAML server does not report document symbol support")
self.server.notify.initialized({})
# YAML language server is ready immediately after initialization
log.info("YAML server initialization complete")
self.server_ready.set()
self.completions_available.set()
```
--------------------------------------------------------------------------------
/src/serena/tools/jetbrains_tools.py:
--------------------------------------------------------------------------------
```python
from collections import defaultdict
from typing import Any
from serena.tools import Tool, ToolMarkerOptional, ToolMarkerSymbolicRead
from serena.tools.jetbrains_plugin_client import JetBrainsPluginClient
class JetBrainsFindSymbolTool(Tool, ToolMarkerSymbolicRead, ToolMarkerOptional):
"""
Performs a global (or local) search for symbols using the JetBrains backend
"""
def apply(
self,
name_path_pattern: str,
depth: int = 0,
relative_path: str | None = None,
include_body: bool = False,
search_deps: bool = False,
max_answer_chars: int = -1,
) -> str:
"""
Retrieves information on all symbols/code entities (classes, methods, etc.) based on the given name path pattern.
The returned symbol information can be used for edits or further queries.
Specify `depth > 0` to retrieve children (e.g., methods of a class).
A name path is a path in the symbol tree *within a source file*.
For example, the method `my_method` defined in class `MyClass` would have the name path `MyClass/my_method`.
If a symbol is overloaded (e.g., in Java), a 0-based index is appended (e.g. "MyClass/my_method[0]") to
uniquely identify it.
To search for a symbol, you provide a name path pattern that is used to match against name paths.
It can be
* a simple name (e.g. "method"), which will match any symbol with that name
* a relative path like "class/method", which will match any symbol with that name path suffix
* an absolute name path "/class/method" (absolute name path), which requires an exact match of the full name path within the source file.
Append an index `[i]` to match a specific overload only, e.g. "MyClass/my_method[1]".
:param name_path_pattern: the name path matching pattern (see above)
:param depth: depth up to which descendants shall be retrieved (e.g. use 1 to also retrieve immediate children;
for the case where the symbol is a class, this will return its methods).
Default 0.
:param relative_path: Optional. Restrict search to this file or directory. If None, searches entire codebase.
If a directory is passed, the search will be restricted to the files in that directory.
If a file is passed, the search will be restricted to that file.
If you have some knowledge about the codebase, you should use this parameter, as it will significantly
speed up the search as well as reduce the number of results.
:param include_body: If True, include the symbol's source code. Use judiciously.
:param search_deps: If True, also search in project dependencies (e.g., libraries).
:param max_answer_chars: max characters for the JSON result. If exceeded, no content is returned.
-1 means the default value from the config will be used.
:return: JSON string: a list of symbols (with locations) matching the name.
"""
if relative_path == ".":
relative_path = None
with JetBrainsPluginClient.from_project(self.project) as client:
response_dict = client.find_symbol(
name_path=name_path_pattern,
relative_path=relative_path,
depth=depth,
include_body=include_body,
search_deps=search_deps,
)
result = self._to_json(response_dict)
return self._limit_length(result, max_answer_chars)
class JetBrainsFindReferencingSymbolsTool(Tool, ToolMarkerSymbolicRead, ToolMarkerOptional):
"""
Finds symbols that reference the given symbol using the JetBrains backend
"""
def apply(
self,
name_path: str,
relative_path: str,
max_answer_chars: int = -1,
) -> str:
"""
Finds symbols that reference the symbol at the given `name_path`.
The result will contain metadata about the referencing symbols.
:param name_path: name path of the symbol for which to find references; matching logic as described in find symbol tool.
:param relative_path: the relative path to the file containing the symbol for which to find references.
Note that here you can't pass a directory but must pass a file.
:param max_answer_chars: max characters for the JSON result. If exceeded, no content is returned. -1 means the
default value from the config will be used.
:return: a list of JSON objects with the symbols referencing the requested symbol
"""
with JetBrainsPluginClient.from_project(self.project) as client:
response_dict = client.find_references(
name_path=name_path,
relative_path=relative_path,
)
result = self._to_json(response_dict)
return self._limit_length(result, max_answer_chars)
class JetBrainsGetSymbolsOverviewTool(Tool, ToolMarkerSymbolicRead, ToolMarkerOptional):
"""
Retrieves an overview of the top-level symbols within a specified file using the JetBrains backend
"""
USE_COMPACT_FORMAT = True
@staticmethod
def _transform_symbols_to_compact_format(symbols: list[dict[str, Any]]) -> dict[str, list]:
"""
Transform symbol overview from verbose format to compact grouped format.
Groups symbols by kind and uses names instead of full symbol objects.
For symbols with children, creates nested dictionaries.
The name_path can be inferred from the hierarchical structure:
- Top-level symbols: name_path = name
- Nested symbols: name_path = parent_name + "/" + name
For example, "convert" under class "ProjectType" has name_path "ProjectType/convert".
"""
result = defaultdict(list)
for symbol in symbols:
kind = symbol.get("type", "Unknown")
name_path = symbol["name_path"]
name = name_path.split("/")[-1]
children = symbol.get("children", [])
if children:
# Symbol has children: create nested dict {name: children_dict}
children_dict = JetBrainsGetSymbolsOverviewTool._transform_symbols_to_compact_format(children)
result[kind].append({name: children_dict})
else:
# Symbol has no children: just add the name
result[kind].append(name)
return result
def apply(
self,
relative_path: str,
depth: int = 0,
max_answer_chars: int = -1,
) -> str:
"""
Gets an overview of the top-level symbols in the given file.
Calling this is often a good idea before more targeted reading, searching or editing operations on the code symbols.
Before requesting a symbol overview, it is usually a good idea to narrow down the scope of the overview
by first understanding the basic directory structure of the repository that you can get from memories
or by using the `list_dir` and `find_file` tools (or similar).
:param relative_path: the relative path to the file to get the overview of
:param depth: depth up to which descendants shall be retrieved (e.g., use 1 to also retrieve immediate children).
:param max_answer_chars: max characters for the JSON result. If exceeded, no content is returned.
-1 means the default value from the config will be used.
:return: a JSON object containing the symbols grouped by kind in a compact format.
"""
with JetBrainsPluginClient.from_project(self.project) as client:
response_dict = client.get_symbols_overview(relative_path=relative_path, depth=depth)
symbols = response_dict["symbols"]
if self.USE_COMPACT_FORMAT:
compact_symbols = self._transform_symbols_to_compact_format(symbols)
result = self._to_json(compact_symbols)
else:
result = self._to_json(symbols)
return self._limit_length(result, max_answer_chars)
```
--------------------------------------------------------------------------------
/test/solidlsp/al/test_al_basic.py:
--------------------------------------------------------------------------------
```python
import os
import pytest
from solidlsp import SolidLanguageServer
from solidlsp.ls_config import Language
from solidlsp.ls_utils import SymbolUtils
from test.conftest import language_tests_enabled
pytestmark = [pytest.mark.al, pytest.mark.skipif(not language_tests_enabled(Language.AL), reason="AL tests are disabled")]
@pytest.mark.al
class TestALLanguageServer:
@pytest.mark.parametrize("language_server", [Language.AL], indirect=True)
def test_find_symbol(self, language_server: SolidLanguageServer) -> None:
"""Test that AL Language Server can find symbols in the test repository."""
symbols = language_server.request_full_symbol_tree()
# Check for table symbols - AL returns full object names like 'Table 50000 "TEST Customer"'
assert SymbolUtils.symbol_tree_contains_name(symbols, 'Table 50000 "TEST Customer"'), "TEST Customer table not found in symbol tree"
# Check for page symbols
assert SymbolUtils.symbol_tree_contains_name(
symbols, 'Page 50001 "TEST Customer Card"'
), "TEST Customer Card page not found in symbol tree"
assert SymbolUtils.symbol_tree_contains_name(
symbols, 'Page 50002 "TEST Customer List"'
), "TEST Customer List page not found in symbol tree"
# Check for codeunit symbols
assert SymbolUtils.symbol_tree_contains_name(symbols, "Codeunit 50000 CustomerMgt"), "CustomerMgt codeunit not found in symbol tree"
assert SymbolUtils.symbol_tree_contains_name(
symbols, "Codeunit 50001 PaymentProcessorImpl"
), "PaymentProcessorImpl codeunit not found in symbol tree"
# Check for enum symbol
assert SymbolUtils.symbol_tree_contains_name(symbols, "Enum 50000 CustomerType"), "CustomerType enum not found in symbol tree"
# Check for interface symbol
assert SymbolUtils.symbol_tree_contains_name(
symbols, "Interface IPaymentProcessor"
), "IPaymentProcessor interface not found in symbol tree"
@pytest.mark.parametrize("language_server", [Language.AL], indirect=True)
def test_find_table_fields(self, language_server: SolidLanguageServer) -> None:
"""Test that AL Language Server can find fields within a table."""
file_path = os.path.join("src", "Tables", "Customer.Table.al")
symbols = language_server.request_document_symbols(file_path).get_all_symbols_and_roots()
# AL tables should have their fields as child symbols
customer_table = None
_all_symbols, root_symbols = symbols
for sym in root_symbols:
if "TEST Customer" in sym.get("name", ""):
customer_table = sym
break
assert customer_table is not None, "Could not find TEST Customer table symbol"
# Check for field symbols (AL nests fields under a "fields" group)
if "children" in customer_table:
# Find the fields group
fields_group = None
for child in customer_table.get("children", []):
if child.get("name") == "fields":
fields_group = child
break
assert fields_group is not None, "Fields group not found in Customer table"
# Check actual field names
if "children" in fields_group:
field_names = [child.get("name", "") for child in fields_group.get("children", [])]
assert any("Name" in name for name in field_names), f"Name field not found. Fields: {field_names}"
assert any("Balance" in name for name in field_names), f"Balance field not found. Fields: {field_names}"
@pytest.mark.parametrize("language_server", [Language.AL], indirect=True)
def test_find_procedures(self, language_server: SolidLanguageServer) -> None:
"""Test that AL Language Server can find procedures in codeunits."""
file_path = os.path.join("src", "Codeunits", "CustomerMgt.Codeunit.al")
symbols = language_server.request_document_symbols(file_path).get_all_symbols_and_roots()
# Find the codeunit symbol - AL returns 'Codeunit 50000 CustomerMgt'
codeunit_symbol = None
_all_symbols, root_symbols = symbols
for sym in root_symbols:
if "CustomerMgt" in sym.get("name", ""):
codeunit_symbol = sym
break
assert codeunit_symbol is not None, "Could not find CustomerMgt codeunit symbol"
# Check for procedure symbols (if hierarchical)
if "children" in codeunit_symbol:
procedure_names = [child.get("name", "") for child in codeunit_symbol.get("children", [])]
assert any("CreateCustomer" in name for name in procedure_names), "CreateCustomer procedure not found"
# Note: UpdateCustomerBalance doesn't exist in our test repo, check for actual procedures
assert any("TestNoSeries" in name for name in procedure_names), "TestNoSeries procedure not found"
@pytest.mark.parametrize("language_server", [Language.AL], indirect=True)
def test_find_referencing_symbols(self, language_server: SolidLanguageServer) -> None:
"""Test that AL Language Server can find references to symbols."""
# Find references to the Customer table from the CustomerMgt codeunit
table_file = os.path.join("src", "Tables", "Customer.Table.al")
symbols = language_server.request_document_symbols(table_file).get_all_symbols_and_roots()
# Find the Customer table symbol
customer_symbol = None
_all_symbols, root_symbols = symbols
for sym in root_symbols:
if "TEST Customer" in sym.get("name", ""):
customer_symbol = sym
break
if customer_symbol and "selectionRange" in customer_symbol:
sel_start = customer_symbol["selectionRange"]["start"]
refs = language_server.request_references(table_file, sel_start["line"], sel_start["character"])
# The Customer table should be referenced in CustomerMgt.Codeunit.al
assert any(
"CustomerMgt.Codeunit.al" in ref.get("relativePath", "") for ref in refs
), "Customer table should be referenced in CustomerMgt.Codeunit.al"
# It should also be referenced in CustomerCard.Page.al
assert any(
"CustomerCard.Page.al" in ref.get("relativePath", "") for ref in refs
), "Customer table should be referenced in CustomerCard.Page.al"
@pytest.mark.parametrize("language_server", [Language.AL], indirect=True)
def test_cross_file_symbols(self, language_server: SolidLanguageServer) -> None:
"""Test that AL Language Server can handle cross-file symbol relationships."""
# Get all symbols to verify cross-file visibility
symbols = language_server.request_full_symbol_tree()
# Count how many AL-specific symbols we found
al_symbols = []
def collect_symbols(syms):
for sym in syms:
if isinstance(sym, dict):
name = sym.get("name", "")
# Look for AL object names (Table, Page, Codeunit, etc.)
if any(keyword in name for keyword in ["Table", "Page", "Codeunit", "Enum", "Interface"]):
al_symbols.append(name)
if "children" in sym:
collect_symbols(sym["children"])
collect_symbols(symbols)
# We should find symbols from multiple files
assert len(al_symbols) >= 5, f"Expected at least 5 AL object symbols, found {len(al_symbols)}: {al_symbols}"
# Verify we have symbols from different AL object types
has_table = any("Table" in s for s in al_symbols)
has_page = any("Page" in s for s in al_symbols)
has_codeunit = any("Codeunit" in s for s in al_symbols)
assert has_table, f"No Table symbols found in: {al_symbols}"
assert has_page, f"No Page symbols found in: {al_symbols}"
assert has_codeunit, f"No Codeunit symbols found in: {al_symbols}"
```
--------------------------------------------------------------------------------
/docs/_config.yml:
--------------------------------------------------------------------------------
```yaml
# Book settings
# Learn more at https://jupyterbook.org/customize/config.html
#######################################################################################
# A default configuration that will be loaded for all jupyter books
# Users are expected to override these values in their own `_config.yml` file.
# This is also the "master list" of all allowed keys and values.
#######################################################################################
# Book settings
title : Serena Documentation # The title of the book. Will be placed in the left navbar.
author : Oraios AI & Oraios Software # The author of the book
copyright : "2025 by Serena contributors" # Copyright year to be placed in the footer
# Patterns to skip when building the book. Can be glob-style (e.g. "*skip.ipynb")
exclude_patterns : ['**.ipynb_checkpoints', '.DS_Store', 'Thumbs.db', '_build', 'jupyter_execute', '.jupyter_cache', '.pytest_cache', 'docs/autogen_rst.py', 'docs/create_toc.py']
# Auto-exclude files not in the toc
only_build_toc_files : true
#######################################################################################
# Execution settings
execute:
# NOTE: Notebooks are not executed, because test_notebooks.py executes them and stores them with outputs in the docs/ folder
# NOTE: If changed, repeat below in `nb_execution_mode`.
execute_notebooks : "off" # Whether to execute notebooks at build time. Must be one of ("auto", "force", "cache", "off")
cache : "" # A path to the jupyter cache that will be used to store execution artifacts. Defaults to `_build/.jupyter_cache/`
exclude_patterns : [] # A list of patterns to *skip* in execution (e.g. a notebook that takes a really long time)
timeout : 1000 # The maximum time (in seconds) each notebook cell is allowed to run.
run_in_temp : false # If `True`, then a temporary directory will be created and used as the command working directory (cwd),
# otherwise the notebook's parent directory will be the cwd.
allow_errors : true # If `False`, when a code cell raises an error the execution is stopped, otherwise all cells are always run.
stderr_output : show # One of 'show', 'remove', 'remove-warn', 'warn', 'error', 'severe'
#######################################################################################
# Parse and render settings
parse:
myst_enable_extensions: # default extensions to enable in the myst parser. See https://myst-parser.readthedocs.io/en/latest/using/syntax-optional.html
- amsmath
- colon_fence
# - deflist
- dollarmath
# - html_admonition
# - html_image
- linkify
# - replacements
# - smartquotes
- substitution
- tasklist
- html_admonition
- html_image
myst_url_schemes: [ mailto, http, https ] # URI schemes that will be recognised as external URLs in Markdown links
myst_dmath_double_inline: true # Allow display math ($$) within an inline context
#######################################################################################
# HTML-specific settings
html:
favicon : "../src/serena/resources/dashboard/serena-icon-32.png"
use_edit_page_button : false # Whether to add an "edit this page" button to pages. If `true`, repository information in repository: must be filled in
use_repository_button : false # Whether to add a link to your repository button
use_issues_button : false # Whether to add an "open an issue" button
use_multitoc_numbering : true # Continuous numbering across parts/chapters
use_darkmode_button : false
extra_footer : ""
home_page_in_navbar : true # Whether to include your home page in the left Navigation Bar
baseurl : "https://oraios.github.io/serena/"
comments:
hypothesis : false
utterances : false
announcement : "" # A banner announcement at the top of the site.
#######################################################################################
# LaTeX-specific settings
latex:
latex_engine : pdflatex # one of 'pdflatex', 'xelatex' (recommended for unicode), 'luatex', 'platex', 'uplatex'
use_jupyterbook_latex : true # use sphinx-jupyterbook-latex for pdf builds as default
targetname : book.tex
# Add a bibtex file so that we can create citations
#bibtex_bibfiles:
# - refs.bib
#######################################################################################
# Launch button settings
launch_buttons:
notebook_interface : classic # The interface interactive links will activate ["classic", "jupyterlab"]
binderhub_url : "" # The URL of the BinderHub (e.g., https://mybinder.org)
jupyterhub_url : "" # The URL of the JupyterHub (e.g., https://datahub.berkeley.edu)
thebe : false # Add a thebe button to pages (requires the repository to run on Binder)
colab_url : "https://colab.research.google.com"
repository:
url : https://github.com/oraios/serena # The URL to your book's repository
path_to_book : docs # A path to your book's folder, relative to the repository root.
branch : main # Which branch of the repository should be used when creating links
#######################################################################################
# Advanced and power-user settings
sphinx:
extra_extensions :
- sphinx.ext.autodoc
- sphinx.ext.viewcode
- sphinx_toolbox.more_autodoc.sourcelink
#- sphinxcontrib.spelling
local_extensions : # A list of local extensions to load by sphinx specified by "name: path" items
recursive_update : false # A boolean indicating whether to overwrite the Sphinx config (true) or recursively update (false)
config : # key-value pairs to directly over-ride the Sphinx configuration
master_doc: "01-about/000_intro.md"
html_theme_options:
logo:
image_light: ../resources/serena-logo.svg
image_dark: ../resources/serena-logo-dark-mode.svg
autodoc_typehints_format: "short"
autodoc_member_order: "bysource"
autoclass_content: "both"
autodoc_default_options:
show-inheritance: True
autodoc_show_sourcelink: True
add_module_names: False
github_username: oraios
github_repository: serena
nb_execution_mode: "off"
nb_merge_streams: True # This is important for cell outputs to appear as single blocks rather than one block per line
python_use_unqualified_type_names: True
nb_mime_priority_overrides: [
[ 'html', 'application/vnd.jupyter.widget-view+json', 10 ],
[ 'html', 'application/javascript', 20 ],
[ 'html', 'text/html', 30 ],
[ 'html', 'text/latex', 40 ],
[ 'html', 'image/svg+xml', 50 ],
[ 'html', 'image/png', 60 ],
[ 'html', 'image/jpeg', 70 ],
[ 'html', 'text/markdown', 80 ],
[ 'html', 'text/plain', 90 ],
[ 'spelling', 'application/vnd.jupyter.widget-view+json', 10 ],
[ 'spelling', 'application/javascript', 20 ],
[ 'spelling', 'text/html', 30 ],
[ 'spelling', 'text/latex', 40 ],
[ 'spelling', 'image/svg+xml', 50 ],
[ 'spelling', 'image/png', 60 ],
[ 'spelling', 'image/jpeg', 70 ],
[ 'spelling', 'text/markdown', 80 ],
[ 'spelling', 'text/plain', 90 ],
]
mathjax_path: https://cdn.jsdelivr.net/npm/mathjax@3/es5/tex-mml-chtml.js
mathjax3_config:
loader: { load: [ '[tex]/configmacros' ] }
tex:
packages: { '[+]': [ 'configmacros' ] }
macros:
vect: ["{\\mathbf{\\boldsymbol{#1}} }", 1]
E: "{\\mathbb{E}}"
P: "{\\mathbb{P}}"
R: "{\\mathbb{R}}"
abs: ["{\\left| #1 \\right|}", 1]
simpl: ["{\\Delta^{#1} }", 1]
amax: "{\\text{argmax}}"
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[build-system]
build-backend = "hatchling.build"
requires = ["hatchling"]
[project]
name = "serena-agent"
version = "0.1.4"
description = ""
authors = [{ name = "Oraios AI", email = "[email protected]" }]
readme = "README.md"
requires-python = ">=3.11, <3.12"
classifiers = [
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3.11",
]
dependencies = [
"requests>=2.32.3,<3",
"pyright>=1.1.396,<2",
"fortls>=3.2.2",
"overrides>=7.7.0,<8",
"python-dotenv>=1.0.0, <2",
"mcp==1.23.0",
"flask>=3.0.0",
"sensai-utils>=1.5.0",
"pydantic>=2.10.6",
"types-pyyaml>=6.0.12.20241230",
"pyyaml>=6.0.2",
"ruamel.yaml==0.18.14",
"jinja2>=3.1.6",
"dotenv>=0.9.9",
"pathspec>=0.12.1",
"psutil>=7.0.0",
"docstring_parser>=0.16",
"joblib>=1.5.1",
"tqdm>=4.67.1",
"tiktoken>=0.9.0",
"anthropic>=0.54.0",
]
[[tool.uv.index]]
name = "testpypi"
url = "https://test.pypi.org/simple/"
publish-url = "https://test.pypi.org/legacy/"
explicit = true
[project.scripts]
serena = "serena.cli:top_level"
serena-mcp-server = "serena.cli:start_mcp_server"
index-project = "serena.cli:index_project" # deprecated
[project.license]
text = "MIT"
[project.optional-dependencies]
dev = [
"black[jupyter]>=23.7.0",
"jinja2",
# In version 1.0.4 we get a NoneType error related to some config conversion (yml_analytics is None and should be a list)
"mypy>=1.16.1",
"poethepoet>=0.20.0",
"pytest>=8.0.2",
"pytest-xdist>=3.5.0",
"ruff>=0.0.285",
"toml-sort>=0.24.2",
"types-pyyaml>=6.0.12.20241230",
"syrupy>=4.9.1",
"types-requests>=2.32.4.20241230",
# docs
"sphinx>=7,<8",
"sphinx_rtd_theme>=0.5.1",
"sphinx-toolbox==3.7.0",
"jupyter-book>=1,<2",
"nbsphinx",
"pyinstrument",
"pytest-timeout>=2.4.0",
]
agno = ["agno>=2.2.1", "sqlalchemy>=2.0.40"]
google = ["google-genai>=1.8.0"]
[project.urls]
Homepage = "https://github.com/oraios/serena"
[tool.hatch.build.targets.wheel]
packages = ["src/serena", "src/interprompt", "src/solidlsp"]
[tool.black]
line-length = 140
target-version = ["py311"]
exclude = '''
/(
src/solidlsp/language_servers/.*/static|src/multilspy
)/
'''
[tool.doc8]
max-line-length = 1000
[tool.mypy]
allow_redefinition = true
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_untyped_defs = true
ignore_missing_imports = true
no_implicit_optional = true
pretty = true
show_error_codes = true
show_error_context = true
show_traceback = true
strict_equality = true
strict_optional = true
warn_no_return = true
warn_redundant_casts = true
warn_unreachable = true
warn_unused_configs = true
warn_unused_ignores = false
exclude = "^build/|^docs/"
[tool.poe.env]
PYDEVD_DISABLE_FILE_VALIDATION = "1"
[tool.poe.tasks]
# Uses PYTEST_MARKERS env var for default markers
# For custom markers, one can either adjust the env var or just use -m option in the command line,
# as the second -m option will override the first one.
test = "pytest test -vv"
_black_check = "black --check src scripts test"
_ruff_check = "ruff check src scripts test"
_black_format = "black src scripts test"
_ruff_format = "ruff check --fix src scripts test"
lint = ["_black_check", "_ruff_check"]
format = ["_ruff_format", "_black_format"]
_mypy = "mypy src/serena src/solidlsp"
type-check = ["_mypy"]
# docs
_autogen_rst = "python docs/autogen_rst.py"
_sphinx_build = "sphinx-build -b html docs docs/_build -W --keep-going"
_jb_generate_toc = "python docs/create_toc.py"
_jb_generate_config = "jupyter-book config sphinx docs/"
doc-clean = "rm -rf docs/_build docs/03_api"
doc-generate-files = ["_autogen_rst", "_jb_generate_toc", "_jb_generate_config"]
doc-build = ["doc-clean", "doc-generate-files", "_sphinx_build"]
[tool.ruff]
target-version = "py311"
line-length = 140
exclude = ["src/solidlsp/language_servers/**/static", "src/multilspy"]
[tool.ruff.format]
quote-style = "double"
indent-style = "space"
line-ending = "auto"
skip-magic-trailing-comma = false
docstring-code-format = true
[tool.ruff.lint]
select = [
"ASYNC",
"B",
"C4",
"C90",
"COM",
"D",
"DTZ",
"E",
"F",
"FLY",
"G",
"I",
"ISC",
"PIE",
"PLC",
"PLE",
"PLW",
"RET",
"RUF",
"RSE",
"SIM",
"TID",
"UP",
"W",
"YTT",
]
ignore = [
"PLC0415",
"RUF002",
"RUF005",
"RUF059",
"SIM118",
"SIM108",
"E501",
"E741",
"B008",
"B011",
"B028",
"D100",
"D101",
"D102",
"D103",
"D104",
"D105",
"D107",
"D200",
"D203",
"D213",
"D401",
"D402",
"DTZ005",
"E402",
"E501",
"E701",
"E731",
"C408",
"E203",
"G004",
"RET505",
"D106",
"D205",
"D212",
"PLW2901",
"B027",
"D404",
"D407",
"D408",
"D409",
"D400",
"D415",
"COM812",
"RET503",
"RET504",
"F403",
"F405",
"C401",
"C901",
"ASYNC230",
"ISC003",
"B024",
"B007",
"SIM102",
"W291",
"W293",
"B009",
"SIM103", # forbids multiple returns
"SIM110", # requires use of any(...) instead of for-loop
"G001", # forbids str.format in log statements
"E722", # forbids unspecific except clause
"SIM105", # forbids empty/general except clause
"SIM113", # wants to enforce use of enumerate
"E712", # forbids equality comparison with True/False
"UP007", # forbids some uses of Union
"TID252", # forbids relative imports
"B904", # forces use of raise from other_exception
"RUF012", # forbids mutable attributes as ClassVar
"SIM117", # forbids nested with statements
"C400", # wants to unnecessarily force use of list comprehension
"UP037", # can incorrectly (!) convert quoted type to unquoted type, causing an error
"UP045", # imposes T | None instead of Optional[T]
"UP031", # forbids % operator to format strings
]
unfixable = ["F841", "F601", "F602", "B018"]
extend-fixable = ["F401", "B905", "W291"]
[tool.ruff.lint.mccabe]
max-complexity = 20
[tool.ruff.lint.per-file-ignores]
"tests/**" = ["D103"]
"scripts/**" = ["D103"]
[tool.pytest.ini_options]
addopts = "--snapshot-patch-pycharm-diff"
markers = [
"clojure: language server running for Clojure",
"python: language server running for Python",
"go: language server running for Go",
"java: language server running for Java",
"kotlin: language server running for kotlin",
"groovy: language server running for Groovy",
"rust: language server running for Rust",
"typescript: language server running for TypeScript",
"vue: language server running for Vue (uses TypeScript LSP)",
"php: language server running for PHP",
"perl: language server running for Perl",
"csharp: language server running for C#",
"elixir: language server running for Elixir",
"elm: language server running for Elm",
"terraform: language server running for Terraform",
"swift: language server running for Swift",
"bash: language server running for Bash",
"r: language server running for R",
"snapshot: snapshot tests for symbolic editing operations",
"ruby: language server running for Ruby (uses ruby-lsp)",
"zig: language server running for Zig",
"lua: language server running for Lua",
"nix: language server running for Nix",
"dart: language server running for Dart",
"erlang: language server running for Erlang",
"scala: language server running for Scala",
"al: language server running for AL (Microsoft Dynamics 365 Business Central)",
"fsharp: language server running for F#",
"rego: language server running for Rego",
"markdown: language server running for Markdown",
"julia: Julia language server tests",
"fortran: language server running for Fortran",
"haskell: Haskell language server tests",
"yaml: language server running for YAML",
"powershell: language server running for PowerShell",
"pascal: language server running for Pascal (Free Pascal/Lazarus)",
"slow: tests that require additional Expert instances and have long startup times (~60-90s each)",
"toml: language server running for TOML",
"matlab: language server running for MATLAB (requires MATLAB R2021b+)",
]
[tool.codespell]
# Ref: https://github.com/codespell-project/codespell#using-a-config-file
skip = '.git*,*.svg,*.lock,*.min.*'
check-hidden = true
# ignore-regex = ''
ignore-words-list = 'paket'
```
--------------------------------------------------------------------------------
/docs/02-usage/050_configuration.md:
--------------------------------------------------------------------------------
```markdown
# Configuration
Serena is very flexible in terms of configuration. While for most users, the default configurations will work,
you can fully adjust it to your needs by editing a few yaml files. You can disable tools, change Serena's instructions
(what we denote as the `system_prompt`), adjust the output of tools that just provide a prompt, and even adjust tool descriptions.
Serena is configured in four places:
1. The `serena_config.yml` for general settings that apply to all clients and projects.
It is located in your user directory under `.serena/serena_config.yml`.
The file will be auto-generated when you first run Serena.
You can edit it directly or use
```shell
<serena> config edit
```
where `<serena>` is [your way of running Serena](020_running).
The configuration file can also be accessed through [Serena's dashboard](060_dashboard).
2. In the arguments passed to the `start-mcp-server` in your client's config (see below),
which will apply to all sessions started by the respective client. In particular, the [context](contexts) parameter
should be set appropriately for Serena to be best adjusted to existing tools and capabilities of your client.
See for a detailed explanation. You can override all entries from the `serena_config.yml` through command line arguments.
3. In the `.serena/project.yml` file within your project. This will hold project-level configuration that is used whenever
that project is activated. This file will be autogenerated when you first use Serena on that project, but you can also
create it explicitly with `serena project create [options]` (see the [](project-creation-indexing)
for details on available options).
4. Through the context and modes (see below).
After the initial setup, continue with one of the sections below, depending on how you
want to use Serena.
## Modes and Contexts
Serena's behavior and toolset can be adjusted using contexts and modes.
These allow for a high degree of customization to best suit your workflow and the environment Serena is operating in.
(contexts)=
### Contexts
A **context** defines the general environment in which Serena is operating.
It influences the initial system prompt and the set of available tools.
A context is set at startup when launching Serena (e.g., via CLI options for an MCP server or in the agent script) and cannot be changed during an active session.
Serena comes with pre-defined contexts:
* `desktop-app`: Tailored for use with desktop applications like Claude Desktop. This is the default.
The full set of Serena's tools is provided, as the application is assumed to have no prior coding-specific capabilities.
* `claude-code`: Optimized for use with Claude Code, it disables tools that would duplicate Claude Code's built-in capabilities.
* `codex`: Optimized for use with OpenAI Codex.
* `ide`: Generic context for IDE assistants/coding agents, e.g. VSCode, Cursor, or Cline, focusing on augmenting existing capabilities.
Basic file operations and shell execution are assumed to be handled by the assistant's own capabilities.
* `agent`: Designed for scenarios where Serena acts as a more autonomous agent, for example, when used with Agno.
Choose the context that best matches the type of integration you are using.
Find the concrete definitions of the above contexts [here](https://github.com/oraios/serena/tree/main/src/serena/resources/config/contexts).
Note that the contexts `ide` and `claude-code` are **single-project contexts** (defining `single_project: true`).
For such contexts, if a project is provided at startup, the set of tools is limited to those required by the project's
concrete configuration, and other tools are excluded completely, allowing the set of tools to be minimal.
Tools explicitly disabled by the project will not be available at all. Since changing the active project
ceases to be a relevant operation in this case, the project activation tool is disabled.
When launching Serena, specify the context using `--context <context-name>`.
Note that for cases where parameter lists are specified (e.g. Claude Desktop), you must add two parameters to the list.
If you are using a local server (such as Llama.cpp) which requires you to use OpenAI-compatible tool descriptions, use context `oaicompat-agent` instead of `agent`.
You can manage contexts using the `context` command,
<serena> context --help
<serena> context list
<serena> context create <context-name>
<serena> context edit <context-name>
<serena> context delete <context-name>
where `<serena>` is [your way of running Serena](020_running).
(modes)=
### Modes
Modes further refine Serena's behavior for specific types of tasks or interaction styles. Multiple modes can be active simultaneously, allowing you to combine their effects. Modes influence the system prompt and can also alter the set of available tools by excluding certain ones.
Examples of built-in modes include:
* `planning`: Focuses Serena on planning and analysis tasks.
* `editing`: Optimizes Serena for direct code modification tasks.
* `interactive`: Suitable for a conversational, back-and-forth interaction style.
* `one-shot`: Configures Serena for tasks that should be completed in a single response, often used with `planning` for generating reports or initial plans.
* `no-onboarding`: Skips the initial onboarding process if it's not needed for a particular session but retains the memory tools (assuming initial memories were created externally).
* `onboarding`: Focuses on the project onboarding process.
* `no-memories`: Disables all memory tools (and tools building on memories such as onboarding tools)
Find the concrete definitions of these modes [here](https://github.com/oraios/serena/tree/main/src/serena/resources/config/modes).
:::{important}
By default, Serena activates the two modes `interactive` and `editing`.
As soon as you start to specify modes, only the modes you explicitly specify will be active, however.
Therefore, if you want to keep the default modes, you must specify them as well.
For example, to add mode `no-memories` to the default behaviour, specify
```shell
--mode interactive --mode editing --mode no-memories
```
:::
Modes can be set at startup (similar to contexts) but can also be _switched dynamically_ during a session.
You can instruct the LLM to use the `switch_modes` tool to activate a different set of modes (e.g., "Switch to planning and one-shot modes").
When launching Serena, specify modes using `--mode <mode-name>`; multiple modes can be specified, e.g. `--mode planning --mode no-onboarding`.
:::{note}
**Mode Compatibility**: While you can combine modes, some may be semantically incompatible (e.g., `interactive` and `one-shot`).
Serena currently does not prevent incompatible combinations; it is up to the user to choose sensible mode configurations.
:::
You can manage modes using the `mode` command,
<serena> mode --help
<serena> mode list
<serena> mode create <mode-name>
<serena> mode edit <mode-name>
<serena> mode delete <mode-name>
where `<serena>` is [your way of running Serena](020_running).
## Advanced Configuration
For advanced users, Serena's configuration can be further customized.
### Serena Data Directory
The Serena user data directory (where configuration, language server files, logs, etc. are stored) defaults to `~/.serena`.
You can change this location by setting the `SERENA_HOME` environment variable to your desired path.
### Custom Prompts
All of Serena's prompts can be fully customized.
We define prompt as jinja templates in yaml files, and you can inspect our default prompts [here](https://github.com/oraios/serena/tree/main/src/serena/resources/config/prompt_templates).
To override a prompt, simply add a .yml file to the `prompt_templates` folder in your Serena data directory
which defines the prompt with the same name as the default prompt you want to override.
For example, to override the `system_prompt`, you could create a file `~/.serena/prompt_templates/system_prompt.yml` (assuming default Serena data folder location)
with content like:
```yaml
prompts:
system_prompt: |
Whatever you want ...
```
It is advisable to use the default prompt as a starting point and modify it to suit your needs.
```
--------------------------------------------------------------------------------
/src/solidlsp/language_servers/pyright_server.py:
--------------------------------------------------------------------------------
```python
"""
Provides Python specific instantiation of the LanguageServer class. Contains various configurations and settings specific to Python.
"""
import logging
import os
import pathlib
import re
import threading
from typing import cast
from overrides import override
from solidlsp.ls import SolidLanguageServer
from solidlsp.ls_config import LanguageServerConfig
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.settings import SolidLSPSettings
log = logging.getLogger(__name__)
class PyrightServer(SolidLanguageServer):
"""
Provides Python specific instantiation of the LanguageServer class using Pyright.
Contains various configurations and settings specific to Python.
"""
def __init__(self, config: LanguageServerConfig, repository_root_path: str, solidlsp_settings: SolidLSPSettings):
"""
Creates a PyrightServer instance. This class is not meant to be instantiated directly.
Use LanguageServer.create() instead.
"""
super().__init__(
config,
repository_root_path,
ProcessLaunchInfo(cmd="python -m pyright.langserver --stdio", cwd=repository_root_path),
"python",
solidlsp_settings,
)
# Event to signal when initial workspace analysis is complete
self.analysis_complete = threading.Event()
self.found_source_files = False
@override
def is_ignored_dirname(self, dirname: str) -> bool:
return super().is_ignored_dirname(dirname) or dirname in ["venv", "__pycache__"]
@staticmethod
def _get_initialize_params(repository_absolute_path: str) -> InitializeParams:
"""
Returns the initialize params for the Pyright Language Server.
"""
# Create basic initialization parameters
initialize_params = { # type: ignore
"processId": os.getpid(),
"rootPath": repository_absolute_path,
"rootUri": pathlib.Path(repository_absolute_path).as_uri(),
"initializationOptions": {
"exclude": [
"**/__pycache__",
"**/.venv",
"**/.env",
"**/build",
"**/dist",
"**/.pixi",
],
"reportMissingImports": "error",
},
"capabilities": {
"workspace": {
"workspaceEdit": {"documentChanges": True},
"didChangeConfiguration": {"dynamicRegistration": True},
"didChangeWatchedFiles": {"dynamicRegistration": True},
"symbol": {
"dynamicRegistration": True,
"symbolKind": {"valueSet": list(range(1, 27))},
},
"executeCommand": {"dynamicRegistration": True},
},
"textDocument": {
"synchronization": {"dynamicRegistration": True, "willSave": True, "willSaveWaitUntil": True, "didSave": True},
"hover": {"dynamicRegistration": True, "contentFormat": ["markdown", "plaintext"]},
"signatureHelp": {
"dynamicRegistration": True,
"signatureInformation": {
"documentationFormat": ["markdown", "plaintext"],
"parameterInformation": {"labelOffsetSupport": True},
},
},
"definition": {"dynamicRegistration": True},
"references": {"dynamicRegistration": True},
"documentSymbol": {
"dynamicRegistration": True,
"symbolKind": {"valueSet": list(range(1, 27))},
"hierarchicalDocumentSymbolSupport": True,
},
"publishDiagnostics": {"relatedInformation": True},
},
},
"workspaceFolders": [
{"uri": pathlib.Path(repository_absolute_path).as_uri(), "name": os.path.basename(repository_absolute_path)}
],
}
return cast(InitializeParams, initialize_params)
def _start_server(self) -> None:
"""
Starts the Pyright Language Server and waits for initial workspace analysis to complete.
This prevents zombie processes by ensuring Pyright has finished its initial background
tasks before we consider the server ready.
Usage:
```
async with lsp.start_server():
# LanguageServer has been initialized and workspace analysis is complete
await lsp.request_definition(...)
await lsp.request_references(...)
# Shutdown the LanguageServer on exit from scope
# LanguageServer has been shutdown cleanly
```
"""
def execute_client_command_handler(params: dict) -> list:
return []
def do_nothing(params: dict) -> None:
return
def window_log_message(msg: dict) -> None:
"""
Monitor Pyright's log messages to detect when initial analysis is complete.
Pyright logs "Found X source files" when it finishes scanning the workspace.
"""
message_text = msg.get("message", "")
log.info(f"LSP: window/logMessage: {message_text}")
# Look for "Found X source files" which indicates workspace scanning is complete
# Unfortunately, pyright is unreliable and there seems to be no better way
if re.search(r"Found \d+ source files?", message_text):
log.info("Pyright workspace scanning complete")
self.found_source_files = True
self.analysis_complete.set()
self.completions_available.set()
def check_experimental_status(params: dict) -> None:
"""
Also listen for experimental/serverStatus as a backup signal
"""
if params.get("quiescent") == True:
log.info("Received experimental/serverStatus with quiescent=true")
if not self.found_source_files:
self.analysis_complete.set()
self.completions_available.set()
# Set up notification handlers
self.server.on_request("client/registerCapability", do_nothing)
self.server.on_notification("language/status", do_nothing)
self.server.on_notification("window/logMessage", window_log_message)
self.server.on_request("workspace/executeClientCommand", execute_client_command_handler)
self.server.on_notification("$/progress", do_nothing)
self.server.on_notification("textDocument/publishDiagnostics", do_nothing)
self.server.on_notification("language/actionableNotification", do_nothing)
self.server.on_notification("experimental/serverStatus", check_experimental_status)
log.info("Starting pyright-langserver server process")
self.server.start()
# Send proper initialization parameters
initialize_params = self._get_initialize_params(self.repository_root_path)
log.info("Sending initialize request from LSP client to pyright server and awaiting response")
init_response = self.server.send.initialize(initialize_params)
log.info(f"Received initialize response from pyright server: {init_response}")
# Verify that the server supports our required features
assert "textDocumentSync" in init_response["capabilities"]
assert "completionProvider" in init_response["capabilities"]
assert "definitionProvider" in init_response["capabilities"]
# Complete the initialization handshake
self.server.notify.initialized({})
# Wait for Pyright to complete its initial workspace analysis
# This prevents zombie processes by ensuring background tasks finish
log.info("Waiting for Pyright to complete initial workspace analysis...")
if self.analysis_complete.wait(timeout=5.0):
log.info("Pyright initial analysis complete, server ready")
else:
log.warning("Timeout waiting for Pyright analysis completion, proceeding anyway")
# Fallback: assume analysis is complete after timeout
self.analysis_complete.set()
self.completions_available.set()
```
--------------------------------------------------------------------------------
/src/solidlsp/language_servers/intelephense.py:
--------------------------------------------------------------------------------
```python
"""
Provides PHP specific instantiation of the LanguageServer class using Intelephense.
"""
import logging
import os
import pathlib
import shutil
from time import sleep
from overrides import override
from solidlsp.ls import SolidLanguageServer
from solidlsp.ls_config import LanguageServerConfig
from solidlsp.ls_utils import PlatformId, PlatformUtils
from solidlsp.lsp_protocol_handler.lsp_types import Definition, DefinitionParams, InitializeParams, LocationLink
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.settings import SolidLSPSettings
from ..lsp_protocol_handler import lsp_types
from .common import RuntimeDependency, RuntimeDependencyCollection
log = logging.getLogger(__name__)
class Intelephense(SolidLanguageServer):
"""
Provides PHP specific instantiation of the LanguageServer class using Intelephense.
You can pass the following entries in ls_specific_settings["php"]:
- maxMemory: sets intelephense.maxMemory
- maxFileSize: sets intelephense.files.maxSize
- ignore_vendor: whether or ignore directories named "vendor" (default: true)
"""
@override
def is_ignored_dirname(self, dirname: str) -> bool:
return super().is_ignored_dirname(dirname) or dirname in self._ignored_dirnames
@classmethod
def _setup_runtime_dependencies(cls, solidlsp_settings: SolidLSPSettings) -> list[str]:
"""
Setup runtime dependencies for Intelephense and return the command to start the server.
"""
platform_id = PlatformUtils.get_platform_id()
valid_platforms = [
PlatformId.LINUX_x64,
PlatformId.LINUX_arm64,
PlatformId.OSX,
PlatformId.OSX_x64,
PlatformId.OSX_arm64,
PlatformId.WIN_x64,
PlatformId.WIN_arm64,
]
assert platform_id in valid_platforms, f"Platform {platform_id} is not supported by {cls.__name__} at the moment"
# Verify both node and npm are installed
is_node_installed = shutil.which("node") is not None
assert is_node_installed, "node is not installed or isn't in PATH. Please install NodeJS and try again."
is_npm_installed = shutil.which("npm") is not None
assert is_npm_installed, "npm is not installed or isn't in PATH. Please install npm and try again."
# Install intelephense if not already installed
intelephense_ls_dir = os.path.join(cls.ls_resources_dir(solidlsp_settings), "php-lsp")
os.makedirs(intelephense_ls_dir, exist_ok=True)
intelephense_executable_path = os.path.join(intelephense_ls_dir, "node_modules", ".bin", "intelephense")
if not os.path.exists(intelephense_executable_path):
deps = RuntimeDependencyCollection(
[
RuntimeDependency(
id="intelephense",
command="npm install --prefix ./ [email protected]",
platform_id="any",
)
]
)
deps.install(intelephense_ls_dir)
assert os.path.exists(
intelephense_executable_path
), f"intelephense executable not found at {intelephense_executable_path}, something went wrong."
return [intelephense_executable_path, "--stdio"]
def __init__(self, config: LanguageServerConfig, repository_root_path: str, solidlsp_settings: SolidLSPSettings):
# Setup runtime dependencies before initializing
intelephense_cmd = self._setup_runtime_dependencies(solidlsp_settings)
super().__init__(
config, repository_root_path, ProcessLaunchInfo(cmd=intelephense_cmd, cwd=repository_root_path), "php", solidlsp_settings
)
self.request_id = 0
# For PHP projects, we should ignore:
# - node_modules: if the project has JavaScript components
# - cache: commonly used for caching
# - (configurable) vendor: third-party dependencies <managed by Composer
self._ignored_dirnames = {"node_modules", "cache"}
if self._custom_settings.get("ignore_vendor", True):
self._ignored_dirnames.add("vendor")
log.info(f"Ignoring the following directories for PHP projects: {', '.join(sorted(self._ignored_dirnames))}")
def _get_initialize_params(self, repository_absolute_path: str) -> InitializeParams:
"""
Returns the initialization params for the Intelephense Language Server.
"""
root_uri = pathlib.Path(repository_absolute_path).as_uri()
initialize_params = {
"locale": "en",
"capabilities": {
"textDocument": {
"synchronization": {"didSave": True, "dynamicRegistration": True},
"definition": {"dynamicRegistration": True},
},
"workspace": {"workspaceFolders": True, "didChangeConfiguration": {"dynamicRegistration": True}},
},
"processId": os.getpid(),
"rootPath": repository_absolute_path,
"rootUri": root_uri,
"workspaceFolders": [
{
"uri": root_uri,
"name": os.path.basename(repository_absolute_path),
}
],
}
initialization_options = {}
# Add license key if provided via environment variable
license_key = os.environ.get("INTELEPHENSE_LICENSE_KEY")
if license_key:
initialization_options["licenceKey"] = license_key
max_memory = self._custom_settings.get("maxMemory")
max_file_size = self._custom_settings.get("maxFileSize")
if max_memory is not None:
initialization_options["intelephense.maxMemory"] = max_memory
if max_file_size is not None:
initialization_options["intelephense.files.maxSize"] = max_file_size
initialize_params["initializationOptions"] = initialization_options
return initialize_params # type: ignore
def _start_server(self) -> None:
"""Start Intelephense server process"""
def register_capability_handler(params: dict) -> None:
return
def window_log_message(msg: dict) -> None:
log.info(f"LSP: window/logMessage: {msg}")
def do_nothing(params: dict) -> None:
return
self.server.on_request("client/registerCapability", register_capability_handler)
self.server.on_notification("window/logMessage", window_log_message)
self.server.on_notification("$/progress", do_nothing)
self.server.on_notification("textDocument/publishDiagnostics", do_nothing)
log.info("Starting Intelephense server process")
self.server.start()
initialize_params = self._get_initialize_params(self.repository_root_path)
log.info("Sending initialize request from LSP client to LSP server and awaiting response")
init_response = self.server.send.initialize(initialize_params)
log.info("After sent initialize params")
# Verify server capabilities
assert "textDocumentSync" in init_response["capabilities"]
assert "completionProvider" in init_response["capabilities"]
assert "definitionProvider" in init_response["capabilities"]
self.server.notify.initialized({})
self.completions_available.set()
# Intelephense server is typically ready immediately after initialization
# TODO: This is probably incorrect; the server does send an initialized notification, which we could wait for!
@override
# For some reason, the LS may need longer to process this, so we just retry
def _send_references_request(self, relative_file_path: str, line: int, column: int) -> list[lsp_types.Location] | None:
# TODO: The LS doesn't return references contained in other files if it doesn't sleep. This is
# despite the LS having processed requests already. I don't know what causes this, but sleeping
# one second helps. It may be that sleeping only once is enough but that's hard to reliably test.
# May be related to the time it takes to read the files or something like that.
# The sleeping doesn't seem to be needed on all systems
sleep(1)
return super()._send_references_request(relative_file_path, line, column)
@override
def _send_definition_request(self, definition_params: DefinitionParams) -> Definition | list[LocationLink] | None:
# TODO: same as above, also only a problem if the definition is in another file
sleep(1)
return super()._send_definition_request(definition_params)
```
--------------------------------------------------------------------------------
/src/solidlsp/language_servers/perl_language_server.py:
--------------------------------------------------------------------------------
```python
"""
Provides Perl specific instantiation of the LanguageServer class using Perl::LanguageServer.
Note: Windows is not supported as Nix itself doesn't support Windows natively.
"""
import logging
import os
import pathlib
import subprocess
import time
from typing import Any
from overrides import override
from solidlsp.ls import SolidLanguageServer
from solidlsp.ls_config import LanguageServerConfig
from solidlsp.ls_utils import PlatformId, PlatformUtils
from solidlsp.lsp_protocol_handler.lsp_types import DidChangeConfigurationParams, InitializeParams
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.settings import SolidLSPSettings
log = logging.getLogger(__name__)
class PerlLanguageServer(SolidLanguageServer):
"""
Provides Perl specific instantiation of the LanguageServer class using Perl::LanguageServer.
"""
@staticmethod
def _get_perl_version() -> str | None:
"""Get the installed Perl version or None if not found."""
try:
result = subprocess.run(["perl", "-v"], capture_output=True, text=True, check=False)
if result.returncode == 0:
return result.stdout.strip()
except FileNotFoundError:
return None
return None
@staticmethod
def _get_perl_language_server_version() -> str | None:
"""Get the installed Perl::LanguageServer version or None if not found."""
try:
result = subprocess.run(
["perl", "-MPerl::LanguageServer", "-e", "print $Perl::LanguageServer::VERSION"],
capture_output=True,
text=True,
check=False,
)
if result.returncode == 0:
return result.stdout.strip()
except FileNotFoundError:
return None
return None
@override
def is_ignored_dirname(self, dirname: str) -> bool:
# For Perl projects, we should ignore:
# - blib: build library directory
# - local: local Perl module installation
# - .carton: Carton dependency manager cache
# - vendor: vendored dependencies
# - _build: Module::Build output
return super().is_ignored_dirname(dirname) or dirname in ["blib", "local", ".carton", "vendor", "_build", "cover_db"]
@classmethod
def _setup_runtime_dependencies(cls) -> str:
"""
Check if required Perl runtime dependencies are available.
Raises RuntimeError with helpful message if dependencies are missing.
"""
platform_id = PlatformUtils.get_platform_id()
valid_platforms = [
PlatformId.LINUX_x64,
PlatformId.LINUX_arm64,
PlatformId.OSX,
PlatformId.OSX_x64,
PlatformId.OSX_arm64,
]
if platform_id not in valid_platforms:
raise RuntimeError(f"Platform {platform_id} is not supported for Perl at the moment")
perl_version = cls._get_perl_version()
if not perl_version:
raise RuntimeError(
"Perl is not installed. Please install Perl from https://www.perl.org/get.html and make sure it is added to your PATH."
)
perl_ls_version = cls._get_perl_language_server_version()
if not perl_ls_version:
raise RuntimeError(
"Found a Perl version but Perl::LanguageServer is not installed.\n"
"Please install Perl::LanguageServer: cpanm Perl::LanguageServer\n"
"See: https://metacpan.org/pod/Perl::LanguageServer"
)
return "perl -MPerl::LanguageServer -e 'Perl::LanguageServer::run'"
def __init__(self, config: LanguageServerConfig, repository_root_path: str, solidlsp_settings: SolidLSPSettings):
# Setup runtime dependencies before initializing
perl_ls_cmd = self._setup_runtime_dependencies()
super().__init__(
config, repository_root_path, ProcessLaunchInfo(cmd=perl_ls_cmd, cwd=repository_root_path), "perl", solidlsp_settings
)
self.request_id = 0
@staticmethod
def _get_initialize_params(repository_absolute_path: str) -> InitializeParams:
"""
Returns the initialize params for Perl::LanguageServer.
Based on the expected structure from Perl::LanguageServer::Methods::_rpcreq_initialize.
"""
root_uri = pathlib.Path(repository_absolute_path).as_uri()
initialize_params = {
"processId": os.getpid(),
"rootPath": repository_absolute_path,
"rootUri": root_uri,
"capabilities": {
"textDocument": {
"synchronization": {"didSave": True, "dynamicRegistration": True},
"definition": {"dynamicRegistration": True},
"references": {"dynamicRegistration": True},
"documentSymbol": {"dynamicRegistration": True},
"hover": {"dynamicRegistration": True},
},
"workspace": {
"workspaceFolders": True,
"didChangeConfiguration": {"dynamicRegistration": True},
"symbol": {"dynamicRegistration": True},
},
},
"initializationOptions": {},
"workspaceFolders": [
{
"uri": root_uri,
"name": os.path.basename(repository_absolute_path),
}
],
}
return initialize_params # type: ignore
def _start_server(self) -> None:
"""Start Perl::LanguageServer process"""
def register_capability_handler(params: Any) -> None:
return
def window_log_message(msg: dict) -> None:
log.info(f"LSP: window/logMessage: {msg}")
def do_nothing(params: Any) -> None:
return
def workspace_configuration_handler(params: Any) -> Any:
"""Handle workspace/configuration request from Perl::LanguageServer."""
log.info(f"Received workspace/configuration request: {params}")
perl_config = {
"perlInc": [self.repository_root_path, "."],
"fileFilter": [".pm", ".pl"],
"ignoreDirs": [".git", ".svn", "blib", "local", ".carton", "vendor", "_build", "cover_db"],
}
return [perl_config]
self.server.on_request("client/registerCapability", register_capability_handler)
self.server.on_request("workspace/configuration", workspace_configuration_handler)
self.server.on_notification("window/logMessage", window_log_message)
self.server.on_notification("$/progress", do_nothing)
self.server.on_notification("textDocument/publishDiagnostics", do_nothing)
log.info("Starting Perl::LanguageServer process")
self.server.start()
initialize_params = self._get_initialize_params(self.repository_root_path)
log.info("Sending initialize request from LSP client to LSP server and awaiting response")
init_response = self.server.send.initialize(initialize_params)
log.info(
"After sent initialize params",
)
# Verify server capabilities
assert "textDocumentSync" in init_response["capabilities"]
assert "definitionProvider" in init_response["capabilities"]
assert "referencesProvider" in init_response["capabilities"]
self.server.notify.initialized({})
# Send workspace configuration to Perl::LanguageServer
# Perl::LanguageServer requires didChangeConfiguration to set perlInc, fileFilter, and ignoreDirs
# See: Perl::LanguageServer::Methods::workspace::_rpcnot_didChangeConfiguration
perl_config: DidChangeConfigurationParams = {
"settings": {
"perl": {
"perlInc": [self.repository_root_path, "."],
"fileFilter": [".pm", ".pl"],
"ignoreDirs": [".git", ".svn", "blib", "local", ".carton", "vendor", "_build", "cover_db"],
}
}
}
log.info(f"Sending workspace/didChangeConfiguration notification with config: {perl_config}")
self.server.notify.workspace_did_change_configuration(perl_config)
self.completions_available.set()
# Perl::LanguageServer needs time to index files and resolve cross-file references
# Without this delay, requests for definitions/references may return empty results
settling_time = 0.5
log.info(f"Allowing {settling_time} seconds for Perl::LanguageServer to index files...")
time.sleep(settling_time)
log.info("Perl::LanguageServer settling period complete")
```
--------------------------------------------------------------------------------
/src/solidlsp/language_servers/elm_language_server.py:
--------------------------------------------------------------------------------
```python
"""
Provides Elm specific instantiation of the LanguageServer class. Contains various configurations and settings specific to Elm.
"""
import logging
import os
import pathlib
import shutil
import threading
from overrides import override
from sensai.util.logging import LogTime
from solidlsp.ls import SolidLanguageServer
from solidlsp.ls_config import LanguageServerConfig
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.settings import SolidLSPSettings
from .common import RuntimeDependency, RuntimeDependencyCollection
log = logging.getLogger(__name__)
class ElmLanguageServer(SolidLanguageServer):
"""
Provides Elm specific instantiation of the LanguageServer class. Contains various configurations and settings specific to Elm.
"""
def __init__(self, config: LanguageServerConfig, repository_root_path: str, solidlsp_settings: SolidLSPSettings):
"""
Creates an ElmLanguageServer instance. This class is not meant to be instantiated directly. Use LanguageServer.create() instead.
"""
elm_lsp_executable_path = self._setup_runtime_dependencies(config, solidlsp_settings)
# Resolve ELM_HOME to absolute path if it's set to a relative path
env = {}
elm_home = os.environ.get("ELM_HOME")
if elm_home:
if not os.path.isabs(elm_home):
# Convert relative ELM_HOME to absolute based on repository root
elm_home = os.path.abspath(os.path.join(repository_root_path, elm_home))
env["ELM_HOME"] = elm_home
log.info(f"Using ELM_HOME: {elm_home}")
super().__init__(
config,
repository_root_path,
ProcessLaunchInfo(cmd=elm_lsp_executable_path, cwd=repository_root_path, env=env),
"elm",
solidlsp_settings,
)
self.server_ready = threading.Event()
@override
def is_ignored_dirname(self, dirname: str) -> bool:
return super().is_ignored_dirname(dirname) or dirname in [
"elm-stuff",
"node_modules",
"dist",
"build",
]
@classmethod
def _setup_runtime_dependencies(cls, config: LanguageServerConfig, solidlsp_settings: SolidLSPSettings) -> list[str]:
"""
Setup runtime dependencies for Elm Language Server and return the command to start the server.
"""
# Check if elm-language-server is already installed globally
system_elm_ls = shutil.which("elm-language-server")
if system_elm_ls:
log.info(f"Found system-installed elm-language-server at {system_elm_ls}")
return [system_elm_ls, "--stdio"]
# Verify node and npm are installed
is_node_installed = shutil.which("node") is not None
assert is_node_installed, "node is not installed or isn't in PATH. Please install NodeJS and try again."
is_npm_installed = shutil.which("npm") is not None
assert is_npm_installed, "npm is not installed or isn't in PATH. Please install npm and try again."
deps = RuntimeDependencyCollection(
[
RuntimeDependency(
id="elm-language-server",
description="@elm-tooling/elm-language-server package",
command=["npm", "install", "--prefix", "./", "@elm-tooling/[email protected]"],
platform_id="any",
),
]
)
# Install elm-language-server if not already installed
elm_ls_dir = os.path.join(cls.ls_resources_dir(solidlsp_settings), "elm-lsp")
elm_ls_executable_path = os.path.join(elm_ls_dir, "node_modules", ".bin", "elm-language-server")
if not os.path.exists(elm_ls_executable_path):
log.info(f"Elm Language Server executable not found at {elm_ls_executable_path}. Installing...")
with LogTime("Installation of Elm language server dependencies", logger=log):
deps.install(elm_ls_dir)
if not os.path.exists(elm_ls_executable_path):
raise FileNotFoundError(
f"elm-language-server executable not found at {elm_ls_executable_path}, something went wrong with the installation."
)
return [elm_ls_executable_path, "--stdio"]
@staticmethod
def _get_initialize_params(repository_absolute_path: str) -> InitializeParams:
"""
Returns the initialize params for the Elm Language Server.
"""
root_uri = pathlib.Path(repository_absolute_path).as_uri()
initialize_params = {
"locale": "en",
"capabilities": {
"textDocument": {
"synchronization": {"didSave": True, "dynamicRegistration": True},
"completion": {"dynamicRegistration": True, "completionItem": {"snippetSupport": True}},
"definition": {"dynamicRegistration": True},
"references": {"dynamicRegistration": True},
"documentSymbol": {
"dynamicRegistration": True,
"hierarchicalDocumentSymbolSupport": True,
"symbolKind": {"valueSet": list(range(1, 27))},
},
"hover": {"dynamicRegistration": True, "contentFormat": ["markdown", "plaintext"]},
"codeAction": {"dynamicRegistration": True},
"rename": {"dynamicRegistration": True},
},
"workspace": {
"workspaceFolders": True,
"didChangeConfiguration": {"dynamicRegistration": True},
"symbol": {"dynamicRegistration": True},
},
},
"initializationOptions": {
"elmPath": shutil.which("elm") or "elm",
"elmFormatPath": shutil.which("elm-format") or "elm-format",
"elmTestPath": shutil.which("elm-test") or "elm-test",
"skipInstallPackageConfirmation": True,
"onlyUpdateDiagnosticsOnSave": False,
},
"processId": os.getpid(),
"rootPath": repository_absolute_path,
"rootUri": root_uri,
"workspaceFolders": [
{
"uri": root_uri,
"name": os.path.basename(repository_absolute_path),
}
],
}
return initialize_params # type: ignore[return-value]
def _start_server(self) -> None:
"""
Starts the Elm Language Server, waits for the server to be ready and yields the LanguageServer instance.
"""
workspace_ready = threading.Event()
def do_nothing(params: dict) -> None:
return
def window_log_message(msg: dict) -> None:
log.info(f"LSP: window/logMessage: {msg}")
def on_diagnostics(params: dict) -> None:
# Receiving diagnostics indicates the workspace has been scanned
log.info("LSP: Received diagnostics notification, workspace is ready")
workspace_ready.set()
self.server.on_notification("window/logMessage", window_log_message)
self.server.on_notification("$/progress", do_nothing)
self.server.on_notification("textDocument/publishDiagnostics", on_diagnostics)
log.info("Starting Elm server process")
self.server.start()
initialize_params = self._get_initialize_params(self.repository_root_path)
log.info("Sending initialize request from LSP client to LSP server and awaiting response")
init_response = self.server.send.initialize(initialize_params)
# Elm-specific capability checks
assert "textDocumentSync" in init_response["capabilities"]
assert "completionProvider" in init_response["capabilities"]
assert "definitionProvider" in init_response["capabilities"]
assert "referencesProvider" in init_response["capabilities"]
assert "documentSymbolProvider" in init_response["capabilities"]
self.server.notify.initialized({})
log.info("Elm server initialized, waiting for workspace scan...")
# Wait for workspace to be scanned (indicated by receiving diagnostics)
if workspace_ready.wait(timeout=30.0):
log.info("Elm server workspace scan completed")
else:
log.warning("Timeout waiting for Elm workspace scan, proceeding anyway")
self.server_ready.set()
self.completions_available.set()
log.info("Elm server ready")
@override
def _get_wait_time_for_cross_file_referencing(self) -> float:
return 1.0
```
--------------------------------------------------------------------------------
/docs/02-usage/020_running.md:
--------------------------------------------------------------------------------
```markdown
# Running Serena
Serena is a command-line tool with a variety of sub-commands.
This section describes
* various ways of running Serena
* how to run and configure the most important command, i.e. starting the MCP server
* other useful commands.
## Ways of Running Serena
In the following, we will refer to the command used to run Serena as `<serena>`,
which you should replace with the appropriate command based on your chosen method,
as detailed below.
In general, to get help, append `--help` to the command, i.e.
<serena> --help
<serena> <command> --help
### Using uvx
`uvx` is part of `uv`. It can be used to run the latest version of Serena directly from the repository, without an explicit local installation.
uvx --from git+https://github.com/oraios/serena serena
Explore the CLI to see some of the customization options that serena provides (more info on them below).
### Local Installation
1. Clone the repository and change into it.
```shell
git clone https://github.com/oraios/serena
cd serena
```
2. Run Serena via
```shell
uv run serena
```
when within the serena installation directory.
From other directories, run it with the `--directory` option, i.e.
```shell
uv run --directory /abs/path/to/serena serena
```
:::{note}
Adding the `--directory` option results in the working directory being set to the Serena directory.
As a consequence, you will need to specify paths when using CLI commands that would otherwise operate on the current directory.
:::
(docker)=
### Using Docker
The Docker approach offers several advantages:
* better security isolation for shell command execution
* no need to install language servers and dependencies locally
* consistent environment across different systems
You can run the Serena MCP server directly via Docker as follows,
assuming that the projects you want to work on are all located in `/path/to/your/projects`:
```shell
docker run --rm -i --network host -v /path/to/your/projects:/workspaces/projects ghcr.io/oraios/serena:latest serena
```
This command mounts your projects into the container under `/workspaces/projects`, so when working with projects,
you need to refer to them using the respective path (e.g. `/workspaces/projects/my-project`).
Alternatively, you may use Docker compose with the `compose.yml` file provided in the repository.
See our [advanced Docker usage](https://github.com/oraios/serena/blob/main/DOCKER.md) documentation for more detailed instructions, configuration options, and limitations.
:::{note}
Docker usage is subject to limitations; see the [advanced Docker usage](https://github.com/oraios/serena/blob/main/DOCKER.md) documentation for details.
:::
### Using Nix
If you are using Nix and [have enabled the `nix-command` and `flakes` features](https://nixos.wiki/wiki/flakes), you can run Serena using the following command:
```bash
nix run github:oraios/serena -- <command> [options]
```
You can also install Serena by referencing this repo (`github:oraios/serena`) and using it in your Nix flake. The package is exported as `serena`.
(start-mcp-server)=
## Running the MCP Server
Given your preferred method of running Serena, you can start the MCP server using the `start-mcp-server` command:
<serena> start-mcp-server [options]
Note that no matter how you run the MCP server, Serena will, by default, start a web-based dashboard on localhost that will allow you to inspect
the server's operations, logs, and configuration.
:::{tip}
By default, Serena will use language servers for code understanding and analysis.
With the [Serena JetBrains Plugin](025_jetbrains_plugin), we recently introduced a powerful alternative,
which has several advantages over the language server-based approach.
:::
### Standard I/O Mode
The typical usage involves the client (e.g. Claude Code, Codex or Cursor) running
the MCP server as a subprocess and using the process' stdin/stdout streams to communicate with it.
In order to launch the server, the client thus needs to be provided with the command to run the MCP server.
:::{note}
MCP servers which use stdio as a protocol are somewhat unusual as far as client/server architectures go, as the server
necessarily has to be started by the client in order for communication to take place via the server's standard input/output streams.
In other words, you do not need to start the server yourself. The client application (e.g. Claude Desktop) takes care of this and
therefore needs to be configured with a launch command.
:::
Communication over stdio is the default for the Serena MCP server, so in the simplest
case, you can simply run the `start-mcp-server` command without any additional options.
<serena> start-mcp-server
For example, to run the server in stdio mode via `uvx`, you would run:
uvx --from git+https://github.com/oraios/serena serena start-mcp-server
See the section ["Configuring Your MCP Client"](030_clients) for specific information on how to configure your MCP client (e.g. Claude Code, Codex, Cursor, etc.)
to use such a launch command.
(streamable-http)=
### Streamable HTTP Mode
When using instead the *Streamable HTTP* mode, you control the server lifecycle yourself,
i.e. you start the server and provide the client with the URL to connect to it.
Simply provide `start-mcp-server` with the `--transport streamable-http` option and optionally provide the desired port
via the `--port` option.
<serena> start-mcp-server --transport streamable-http --port <port>
For example, to run the Serena MCP server in streamable HTTP mode on port 9121 using uvx,
you would run
uvx --from git+https://github.com/oraios/serena serena start-mcp-server --transport streamable-http --port 9121
and then configure your client to connect to `http://localhost:9121/mcp`.
Note that while the legacy SSE transport is also supported (via `--transport sse` with corresponding /sse endpoint), its use is discouraged.
(mcp-args)=
### MCP Server Command-Line Arguments
The Serena MCP server supports a wide range of additional command-line options.
Use the command
<serena> start-mcp-server --help
to get a list of all available options.
Some useful options include:
* `--project <path|name>`: specify the project to work on by name or path.
* `--project-from-cwd`: auto-detect the project from current working directory
(looking for a directory containing `.serena/project.yml` or `.git` in parent directories, activating the current directory if none is found);
This option is intended for CLI-based agents like Claude Code, Gemini and Codex, which are typically started from within the project directory
and which do not change directories during their operation.
* `--language-backend JetBrains`: use the Serena JetBrains Plugin as the language backend (overriding the default backend configured in the central configuration)
* `--context <context>`: specify the operation [context](contexts) in which Serena shall operate
* `--mode <mode>`: specify one or more [modes](modes) to enable (can be passed several times)
* `--enable-web-dashboard <true|false>`: enable or disable the web dashboard (enabled by default)
## Other Commands
Serena provides several other commands in addition to `start-mcp-server`,
most of which are related to project setup and configuration.
To get a list of available commands, run:
<serena> --help
To get help on a specific command, run:
<serena> <command> --help
In general, add `--help` to any command or sub-command to get information about its usage and available options.
Here are some examples of commands you might find useful:
```bash
# get help about a sub-command
<serena> tools list --help
# list all available tools
<serena> tools list --all
# get detailed description of a specific tool
<serena> tools description find_symbol
# creating a new Serena project in the current directory
<serena> project create
# creating and immediately indexing a project
<serena> project create --index
# indexing the project in the current directory (auto-creates if needed)
<serena> project index
# run a health check on the project in the current directory
<serena> project health-check
# check if a path is ignored by the project
<serena> project is_ignored_path path/to/check
# edit Serena's configuration file
<serena> config edit
# list available contexts
<serena> context list
# create a new context
<serena> context create my-custom-context
# edit a custom context
<serena> context edit my-custom-context
# list available modes
<serena> mode list
# create a new mode
<serena> mode create my-custom-mode
# edit a custom mode
<serena> mode edit my-custom-mode
# list available prompt definitions
<serena> prompts list
# create an override for internal prompts
<serena> prompts create-override prompt-name
# edit a prompt override
<serena> prompts edit-override prompt-name
```
Explore the full set of commands and options using the CLI itself!
```
--------------------------------------------------------------------------------
/src/solidlsp/language_servers/jedi_server.py:
--------------------------------------------------------------------------------
```python
"""
Provides Python specific instantiation of the LanguageServer class. Contains various configurations and settings specific to Python.
"""
import logging
import os
import pathlib
from typing import cast
from overrides import override
from solidlsp.ls import SolidLanguageServer
from solidlsp.ls_config import LanguageServerConfig
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.settings import SolidLSPSettings
log = logging.getLogger(__name__)
class JediServer(SolidLanguageServer):
"""
Provides Python specific instantiation of the LanguageServer class. Contains various configurations and settings specific to Python.
"""
def __init__(self, config: LanguageServerConfig, repository_root_path: str, solidlsp_settings: SolidLSPSettings):
"""
Creates a JediServer instance. This class is not meant to be instantiated directly. Use LanguageServer.create() instead.
"""
super().__init__(
config,
repository_root_path,
ProcessLaunchInfo(cmd="jedi-language-server", cwd=repository_root_path),
"python",
solidlsp_settings,
)
@override
def is_ignored_dirname(self, dirname: str) -> bool:
return super().is_ignored_dirname(dirname) or dirname in ["venv", "__pycache__"]
@staticmethod
def _get_initialize_params(repository_absolute_path: str) -> InitializeParams:
"""
Returns the initialize params for the Jedi Language Server.
"""
root_uri = pathlib.Path(repository_absolute_path).as_uri()
initialize_params = {
"processId": os.getpid(),
"clientInfo": {"name": "Serena", "version": "0.1.0"},
"locale": "en",
"rootPath": repository_absolute_path,
"rootUri": root_uri,
# Note: this is not necessarily the minimal set of capabilities...
"capabilities": {
"workspace": {
"applyEdit": True,
"workspaceEdit": {
"documentChanges": True,
"resourceOperations": ["create", "rename", "delete"],
"failureHandling": "textOnlyTransactional",
"normalizesLineEndings": True,
"changeAnnotationSupport": {"groupsOnLabel": True},
},
"configuration": True,
"didChangeWatchedFiles": {"dynamicRegistration": True, "relativePatternSupport": True},
"symbol": {
"dynamicRegistration": True,
"symbolKind": {"valueSet": list(range(1, 27))},
"tagSupport": {"valueSet": [1]},
"resolveSupport": {"properties": ["location.range"]},
},
"workspaceFolders": True,
"fileOperations": {
"dynamicRegistration": True,
"didCreate": True,
"didRename": True,
"didDelete": True,
"willCreate": True,
"willRename": True,
"willDelete": True,
},
"inlineValue": {"refreshSupport": True},
"inlayHint": {"refreshSupport": True},
"diagnostics": {"refreshSupport": True},
},
"textDocument": {
"publishDiagnostics": {
"relatedInformation": True,
"versionSupport": False,
"tagSupport": {"valueSet": [1, 2]},
"codeDescriptionSupport": True,
"dataSupport": True,
},
"synchronization": {"dynamicRegistration": True, "willSave": True, "willSaveWaitUntil": True, "didSave": True},
"hover": {"dynamicRegistration": True, "contentFormat": ["markdown", "plaintext"]},
"signatureHelp": {
"dynamicRegistration": True,
"signatureInformation": {
"documentationFormat": ["markdown", "plaintext"],
"parameterInformation": {"labelOffsetSupport": True},
"activeParameterSupport": True,
},
"contextSupport": True,
},
"definition": {"dynamicRegistration": True, "linkSupport": True},
"references": {"dynamicRegistration": True},
"documentHighlight": {"dynamicRegistration": True},
"documentSymbol": {
"dynamicRegistration": True,
"symbolKind": {"valueSet": list(range(1, 27))},
"hierarchicalDocumentSymbolSupport": True,
"tagSupport": {"valueSet": [1]},
"labelSupport": True,
},
"documentLink": {"dynamicRegistration": True, "tooltipSupport": True},
"typeDefinition": {"dynamicRegistration": True, "linkSupport": True},
"implementation": {"dynamicRegistration": True, "linkSupport": True},
"declaration": {"dynamicRegistration": True, "linkSupport": True},
"selectionRange": {"dynamicRegistration": True},
"callHierarchy": {"dynamicRegistration": True},
"linkedEditingRange": {"dynamicRegistration": True},
"typeHierarchy": {"dynamicRegistration": True},
"inlineValue": {"dynamicRegistration": True},
"inlayHint": {
"dynamicRegistration": True,
"resolveSupport": {"properties": ["tooltip", "textEdits", "label.tooltip", "label.location", "label.command"]},
},
"diagnostic": {"dynamicRegistration": True, "relatedDocumentSupport": False},
},
"notebookDocument": {"synchronization": {"dynamicRegistration": True, "executionSummarySupport": True}},
"experimental": {
"serverStatusNotification": True,
"openServerLogs": True,
},
},
# See https://github.com/pappasam/jedi-language-server?tab=readme-ov-file
# We use the default options except for maxSymbols, where 0 means no limit
"initializationOptions": {
"workspace": {
"symbols": {"ignoreFolders": [".nox", ".tox", ".venv", "__pycache__", "venv"], "maxSymbols": 0},
},
},
"trace": "verbose",
"workspaceFolders": [
{
"uri": root_uri,
"name": os.path.basename(repository_absolute_path),
}
],
}
return cast(InitializeParams, initialize_params)
def _start_server(self) -> None:
"""
Starts the JEDI Language Server
"""
def execute_client_command_handler(params: dict) -> list:
return []
def do_nothing(params: dict) -> None:
return
def check_experimental_status(params: dict) -> None:
if params["quiescent"] == True:
self.completions_available.set()
def window_log_message(msg: dict) -> None:
log.info(f"LSP: window/logMessage: {msg}")
self.server.on_request("client/registerCapability", do_nothing)
self.server.on_notification("language/status", do_nothing)
self.server.on_notification("window/logMessage", window_log_message)
self.server.on_request("workspace/executeClientCommand", execute_client_command_handler)
self.server.on_notification("$/progress", do_nothing)
self.server.on_notification("textDocument/publishDiagnostics", do_nothing)
self.server.on_notification("language/actionableNotification", do_nothing)
self.server.on_notification("experimental/serverStatus", check_experimental_status)
log.info("Starting jedi-language-server server process")
self.server.start()
initialize_params = self._get_initialize_params(self.repository_root_path)
log.info("Sending initialize request from LSP client to LSP server and awaiting response")
init_response = self.server.send.initialize(initialize_params)
assert init_response["capabilities"]["textDocumentSync"]["change"] == 2 # type: ignore
assert "completionProvider" in init_response["capabilities"]
assert init_response["capabilities"]["completionProvider"] == {
"triggerCharacters": [".", "'", '"'],
"resolveProvider": True,
}
self.server.notify.initialized({})
```
--------------------------------------------------------------------------------
/src/serena/ls_manager.py:
--------------------------------------------------------------------------------
```python
import logging
import threading
from collections.abc import Iterator
from sensai.util.logging import LogTime
from serena.config.serena_config import SerenaPaths
from serena.constants import SERENA_MANAGED_DIR_NAME
from solidlsp import SolidLanguageServer
from solidlsp.ls_config import Language, LanguageServerConfig
from solidlsp.settings import SolidLSPSettings
log = logging.getLogger(__name__)
class LanguageServerFactory:
def __init__(
self,
project_root: str,
encoding: str,
ignored_patterns: list[str],
ls_timeout: float | None = None,
ls_specific_settings: dict | None = None,
trace_lsp_communication: bool = False,
):
self.project_root = project_root
self.encoding = encoding
self.ignored_patterns = ignored_patterns
self.ls_timeout = ls_timeout
self.ls_specific_settings = ls_specific_settings
self.trace_lsp_communication = trace_lsp_communication
def create_language_server(self, language: Language) -> SolidLanguageServer:
ls_config = LanguageServerConfig(
code_language=language,
ignored_paths=self.ignored_patterns,
trace_lsp_communication=self.trace_lsp_communication,
encoding=self.encoding,
)
log.info(f"Creating language server instance for {self.project_root}, language={language}.")
return SolidLanguageServer.create(
ls_config,
self.project_root,
timeout=self.ls_timeout,
solidlsp_settings=SolidLSPSettings(
solidlsp_dir=SerenaPaths().serena_user_home_dir,
project_data_relative_path=SERENA_MANAGED_DIR_NAME,
ls_specific_settings=self.ls_specific_settings or {},
),
)
class LanguageServerManager:
"""
Manages one or more language servers for a project.
"""
def __init__(
self,
language_servers: dict[Language, SolidLanguageServer],
language_server_factory: LanguageServerFactory | None = None,
) -> None:
"""
:param language_servers: a mapping from language to language server; the servers are assumed to be already started.
The first server in the iteration order is used as the default server.
All servers are assumed to serve the same project root.
:param language_server_factory: factory for language server creation; if None, dynamic (re)creation of language servers
is not supported
"""
self._language_servers = language_servers
self._language_server_factory = language_server_factory
self._default_language_server = next(iter(language_servers.values()))
self._root_path = self._default_language_server.repository_root_path
@staticmethod
def from_languages(languages: list[Language], factory: LanguageServerFactory) -> "LanguageServerManager":
"""
Creates a manager with language servers for the given languages using the given factory.
The language servers are started in parallel threads.
:param languages: the languages for which to spawn language servers
:param factory: the factory for language server creation
:return: the instance
"""
language_servers: dict[Language, SolidLanguageServer] = {}
threads = []
exceptions = {}
lock = threading.Lock()
def start_language_server(language: Language) -> None:
try:
with LogTime(f"Language server startup (language={language.value})"):
language_server = factory.create_language_server(language)
language_server.start()
if not language_server.is_running():
raise RuntimeError(f"Failed to start the language server for language {language.value}")
with lock:
language_servers[language] = language_server
except Exception as e:
log.error(f"Error starting language server for language {language.value}: {e}", exc_info=e)
with lock:
exceptions[language] = e
# start language servers in parallel threads
for language in languages:
thread = threading.Thread(target=start_language_server, args=(language,), name="StartLS:" + language.value)
thread.start()
threads.append(thread)
for thread in threads:
thread.join()
# If any server failed to start up, raise an exception and stop all started language servers
if exceptions:
for ls in language_servers.values():
ls.stop()
failure_messages = "\n".join([f"{lang.value}: {e}" for lang, e in exceptions.items()])
raise Exception(f"Failed to start language servers:\n{failure_messages}")
return LanguageServerManager(language_servers, factory)
def get_root_path(self) -> str:
return self._root_path
def _ensure_functional_ls(self, ls: SolidLanguageServer) -> SolidLanguageServer:
if not ls.is_running():
log.warning(f"Language server for language {ls.language} is not running; restarting ...")
ls = self.restart_language_server(ls.language)
return ls
def get_language_server(self, relative_path: str) -> SolidLanguageServer:
ls: SolidLanguageServer | None = None
if len(self._language_servers) > 1:
for candidate in self._language_servers.values():
if not candidate.is_ignored_path(relative_path, ignore_unsupported_files=True):
ls = candidate
break
if ls is None:
ls = self._default_language_server
return self._ensure_functional_ls(ls)
def _create_and_start_language_server(self, language: Language) -> SolidLanguageServer:
if self._language_server_factory is None:
raise ValueError(f"No language server factory available to create language server for {language}")
language_server = self._language_server_factory.create_language_server(language)
language_server.start()
self._language_servers[language] = language_server
return language_server
def restart_language_server(self, language: Language) -> SolidLanguageServer:
"""
Forces recreation and restart of the language server for the given language.
It is assumed that the language server for the given language is no longer running.
:param language: the language
:return: the newly created language server
"""
if language not in self._language_servers:
raise ValueError(f"No language server for language {language.value} present; cannot restart")
return self._create_and_start_language_server(language)
def add_language_server(self, language: Language) -> SolidLanguageServer:
"""
Dynamically adds a new language server for the given language.
:param language: the language
:param factory: the factory to create the language server
:return: the newly created language server
"""
if language in self._language_servers:
raise ValueError(f"Language server for language {language.value} already present")
return self._create_and_start_language_server(language)
def remove_language_server(self, language: Language, save_cache: bool = False) -> None:
"""
Removes the language server for the given language, stopping it if it is running.
:param language: the language
"""
if language not in self._language_servers:
raise ValueError(f"No language server for language {language.value} present; cannot remove")
ls = self._language_servers.pop(language)
self._stop_language_server(ls, save_cache=save_cache)
@staticmethod
def _stop_language_server(ls: SolidLanguageServer, save_cache: bool = False, timeout: float = 2.0) -> None:
if ls.is_running():
if save_cache:
ls.save_cache()
log.info(f"Stopping language server for language {ls.language} ...")
ls.stop(shutdown_timeout=timeout)
def iter_language_servers(self) -> Iterator[SolidLanguageServer]:
for ls in self._language_servers.values():
yield self._ensure_functional_ls(ls)
def stop_all(self, save_cache: bool = False, timeout: float = 2.0) -> None:
"""
Stops all managed language servers.
:param save_cache: whether to save the cache before stopping
:param timeout: timeout for shutdown of each language server
"""
for ls in self.iter_language_servers():
self._stop_language_server(ls, save_cache=save_cache, timeout=timeout)
def save_all_caches(self) -> None:
"""
Saves the caches of all managed language servers.
"""
for ls in self.iter_language_servers():
if ls.is_running():
ls.save_cache()
```
--------------------------------------------------------------------------------
/src/solidlsp/language_servers/erlang_language_server.py:
--------------------------------------------------------------------------------
```python
"""Erlang Language Server implementation using Erlang LS."""
import logging
import os
import shutil
import subprocess
import threading
import time
from overrides import override
from solidlsp.ls import SolidLanguageServer
from solidlsp.ls_config import LanguageServerConfig
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.settings import SolidLSPSettings
log = logging.getLogger(__name__)
class ErlangLanguageServer(SolidLanguageServer):
"""Language server for Erlang using Erlang LS."""
def __init__(self, config: LanguageServerConfig, repository_root_path: str, solidlsp_settings: SolidLSPSettings):
"""
Creates an ErlangLanguageServer instance. This class is not meant to be instantiated directly.
Use LanguageServer.create() instead.
"""
self.erlang_ls_path = shutil.which("erlang_ls")
if not self.erlang_ls_path:
raise RuntimeError("Erlang LS not found. Install from: https://github.com/erlang-ls/erlang_ls")
if not self._check_erlang_installation():
raise RuntimeError("Erlang/OTP not found. Install from: https://www.erlang.org/downloads")
super().__init__(
config,
repository_root_path,
ProcessLaunchInfo(cmd=[self.erlang_ls_path, "--transport", "stdio"], cwd=repository_root_path),
"erlang",
solidlsp_settings,
)
# Add server readiness tracking like Elixir
self.server_ready = threading.Event()
# Set generous timeout for Erlang LS initialization
self.set_request_timeout(120.0)
def _check_erlang_installation(self) -> bool:
"""Check if Erlang/OTP is available."""
try:
result = subprocess.run(["erl", "-version"], check=False, capture_output=True, text=True, timeout=10)
return result.returncode == 0
except (subprocess.SubprocessError, FileNotFoundError):
return False
@classmethod
def _get_erlang_version(cls) -> str | None:
"""Get the installed Erlang/OTP version or None if not found."""
try:
result = subprocess.run(["erl", "-version"], check=False, capture_output=True, text=True, timeout=10)
if result.returncode == 0:
return result.stderr.strip() # erl -version outputs to stderr
except (subprocess.SubprocessError, FileNotFoundError):
return None
return None
@classmethod
def _check_rebar3_available(cls) -> bool:
"""Check if rebar3 build tool is available."""
try:
result = subprocess.run(["rebar3", "version"], check=False, capture_output=True, text=True, timeout=10)
return result.returncode == 0
except (subprocess.SubprocessError, FileNotFoundError):
return False
def _start_server(self) -> None:
"""Start Erlang LS server process with proper initialization waiting."""
def register_capability_handler(params: dict) -> None:
return
def window_log_message(msg: dict) -> None:
"""Handle window/logMessage notifications from Erlang LS"""
message_text = msg.get("message", "")
log.info(f"LSP: window/logMessage: {message_text}")
# Look for Erlang LS readiness signals
# Common patterns: "Started Erlang LS", "initialized", "ready"
readiness_signals = [
"Started Erlang LS",
"server started",
"initialized",
"ready to serve requests",
"compilation finished",
"indexing complete",
]
message_lower = message_text.lower()
for signal in readiness_signals:
if signal.lower() in message_lower:
log.info(f"Erlang LS readiness signal detected: {message_text}")
self.server_ready.set()
break
def do_nothing(params: dict) -> None:
return
def check_server_ready(params: dict) -> None:
"""Handle $/progress notifications from Erlang LS as fallback."""
value = params.get("value", {})
# Check for initialization completion progress
if value.get("kind") == "end":
message = value.get("message", "")
if any(word in message.lower() for word in ["initialized", "ready", "complete"]):
log.info("Erlang LS initialization progress completed")
# Set as fallback if no window/logMessage was received
if not self.server_ready.is_set():
self.server_ready.set()
# Set up notification handlers
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_notification("window/workDoneProgress/create", do_nothing)
self.server.on_notification("$/workDoneProgress", do_nothing)
self.server.on_notification("textDocument/publishDiagnostics", do_nothing)
log.info("Starting Erlang LS server process")
self.server.start()
# Send initialize request
initialize_params = {
"processId": None,
"rootPath": self.repository_root_path,
"rootUri": f"file://{self.repository_root_path}",
"capabilities": {
"textDocument": {
"synchronization": {"didSave": True},
"completion": {"dynamicRegistration": True},
"definition": {"dynamicRegistration": True},
"references": {"dynamicRegistration": True},
"documentSymbol": {"dynamicRegistration": True},
"hover": {"dynamicRegistration": True},
}
},
}
log.info("Sending initialize request to Erlang LS")
init_response = self.server.send.initialize(initialize_params) # type: ignore[arg-type]
# Verify server capabilities
if "capabilities" in init_response:
log.info(f"Erlang LS capabilities: {list(init_response['capabilities'].keys())}")
self.server.notify.initialized({})
self.completions_available.set()
# Wait for Erlang LS to be ready - adjust timeout based on environment
is_ci = os.getenv("CI") == "true" or os.getenv("GITHUB_ACTIONS") == "true"
is_macos = os.uname().sysname == "Darwin" if hasattr(os, "uname") else False
# macOS in CI can be particularly slow for language server startup
if is_ci and is_macos:
ready_timeout = 240.0 # 4 minutes for macOS CI
env_desc = "macOS CI"
elif is_ci:
ready_timeout = 180.0 # 3 minutes for other CI
env_desc = "CI"
else:
ready_timeout = 60.0 # 1 minute for local
env_desc = "local"
log.info(f"Waiting up to {ready_timeout} seconds for Erlang LS readiness ({env_desc} environment)...")
if self.server_ready.wait(timeout=ready_timeout):
log.info("Erlang LS is ready and available for requests")
# Add settling period for indexing - adjust based on environment
settling_time = 15.0 if is_ci else 5.0
log.info(f"Allowing {settling_time} seconds for Erlang LS indexing to complete...")
time.sleep(settling_time)
log.info("Erlang LS settling period complete")
else:
# Set ready anyway and continue - Erlang LS might not send explicit ready messages
log.warning(f"Erlang LS readiness timeout reached after {ready_timeout}s, proceeding anyway (common in CI)")
self.server_ready.set()
# Still give some time for basic initialization even without explicit readiness signal
basic_settling_time = 20.0 if is_ci else 10.0
log.info(f"Allowing {basic_settling_time} seconds for basic Erlang LS initialization...")
time.sleep(basic_settling_time)
log.info("Basic Erlang LS initialization period complete")
@override
def is_ignored_dirname(self, dirname: str) -> bool:
# For Erlang projects, we should ignore:
# - _build: rebar3 build artifacts
# - deps: dependencies
# - ebin: compiled beam files
# - .rebar3: rebar3 cache
# - logs: log files
# - node_modules: if the project has JavaScript components
return super().is_ignored_dirname(dirname) or dirname in [
"_build",
"deps",
"ebin",
".rebar3",
"logs",
"node_modules",
"_checkouts",
"cover",
]
def is_ignored_filename(self, filename: str) -> bool:
"""Check if a filename should be ignored."""
# Ignore compiled BEAM files
if filename.endswith(".beam"):
return True
# Don't ignore Erlang source files, header files, or configuration files
return False
```
--------------------------------------------------------------------------------
/src/serena/tools/jetbrains_plugin_client.py:
--------------------------------------------------------------------------------
```python
"""
Client for the Serena JetBrains Plugin
"""
import json
import logging
import re
from pathlib import Path
from typing import Any, Optional, Self, TypeVar
import requests
from requests import Response
from sensai.util.string import ToStringMixin
from serena.project import Project
T = TypeVar("T")
log = logging.getLogger(__name__)
class SerenaClientError(Exception):
"""Base exception for Serena client errors."""
class ConnectionError(SerenaClientError):
"""Raised when connection to the service fails."""
class APIError(SerenaClientError):
"""Raised when the API returns an error response."""
class ServerNotFoundError(Exception):
"""Raised when the plugin's service is not found."""
class JetBrainsPluginClient(ToStringMixin):
"""
Python client for the Serena Backend Service.
Provides simple methods to interact with all available endpoints.
"""
BASE_PORT = 0x5EA2
PLUGIN_REQUEST_TIMEOUT = 300
"""
the timeout used for request handling within the plugin (a constant in the plugin)
"""
last_port: int | None = None
def __init__(self, port: int, timeout: int = PLUGIN_REQUEST_TIMEOUT):
self.base_url = f"http://127.0.0.1:{port}"
self.timeout = timeout
self.session = requests.Session()
self.session.headers.update({"Content-Type": "application/json", "Accept": "application/json"})
def _tostring_includes(self) -> list[str]:
return ["base_url", "timeout"]
@classmethod
def from_project(cls, project: Project) -> Self:
resolved_path = Path(project.project_root).resolve()
if cls.last_port is not None:
client = JetBrainsPluginClient(cls.last_port)
if client.matches(resolved_path):
return client
for port in range(cls.BASE_PORT, cls.BASE_PORT + 20):
client = JetBrainsPluginClient(port)
if client.matches(resolved_path):
log.info("Found JetBrains IDE service at port %d for project %s", port, resolved_path)
cls.last_port = port
return client
raise ServerNotFoundError("Found no Serena service in a JetBrains IDE instance for the project at " + str(resolved_path))
@staticmethod
def _normalize_wsl_path(path_str: str) -> Path:
"""
Normalize WSL UNC paths to Linux paths for comparison.
When JetBrains IDE runs on Windows with a project opened from WSL,
it returns paths like `//wsl.localhost/Ubuntu-24.04/home/user/project`
or `//wsl$/Ubuntu/home/user/project`. This method converts such paths
to standard Linux format `/home/user/project` for proper matching.
:param path_str: Path string that may be a WSL UNC path
:return: Normalized Path object
"""
path_str = str(path_str)
# Match //wsl.localhost/<distro>/... or //wsl$/<distro>/...
match = re.match(r"^//wsl(?:\.localhost|\$)/[^/]+(.*)$", path_str, re.IGNORECASE)
if match:
return Path(match.group(1))
return Path(path_str)
def matches(self, resolved_path: Path) -> bool:
try:
plugin_root = self._normalize_wsl_path(self.project_root())
return plugin_root.resolve() == resolved_path
except ConnectionError:
return False
def _make_request(self, method: str, endpoint: str, data: Optional[dict] = None) -> dict[str, Any]:
url = f"{self.base_url}{endpoint}"
response: Response | None = None
try:
if method.upper() == "GET":
response = self.session.get(url, timeout=self.timeout)
elif method.upper() == "POST":
json_data = json.dumps(data) if data else None
response = self.session.post(url, data=json_data, timeout=self.timeout)
else:
raise ValueError(f"Unsupported HTTP method: {method}")
response.raise_for_status()
# Try to parse JSON response
try:
return self._pythonify_response(response.json())
except json.JSONDecodeError:
# If response is not JSON, return raw text
return {"response": response.text}
except requests.exceptions.ConnectionError as e:
raise ConnectionError(f"Failed to connect to Serena service at {url}: {e}")
except requests.exceptions.Timeout as e:
raise ConnectionError(f"Request to {url} timed out: {e}")
except requests.exceptions.HTTPError as e:
if response is not None:
raise APIError(f"API request failed with status {response.status_code}: {response.text}")
raise APIError(f"API request failed with HTTP error: {e}")
except requests.exceptions.RequestException as e:
raise SerenaClientError(f"Request failed: {e}")
@staticmethod
def _pythonify_response(response: T) -> T:
"""
Converts dictionary keys from camelCase to snake_case recursively.
:response: the response in which to convert keys (dictionary or list)
"""
to_snake_case = lambda s: "".join(["_" + c.lower() if c.isupper() else c for c in s])
def convert(x): # type: ignore
if isinstance(x, dict):
return {to_snake_case(k): convert(v) for k, v in x.items()}
elif isinstance(x, list):
return [convert(item) for item in x]
else:
return x
return convert(response)
def project_root(self) -> str:
response = self._make_request("GET", "/status")
return response["project_root"]
def find_symbol(
self,
name_path: str,
relative_path: str | None = None,
include_body: bool = False,
depth: int = 0,
include_location: bool = False,
search_deps: bool = False,
) -> dict[str, Any]:
"""
Finds symbols by name.
:param name_path: the name path to match
:param relative_path: the relative path to which to restrict the search
:param include_body: whether to include symbol body content
:param depth: depth of children to include (0 = no children)
:param include_location: whether to include symbol location information
:param search_deps: whether to also search in dependencies
:return: Dictionary containing 'symbols' list with matching symbols
"""
request_data = {
"namePath": name_path,
"relativePath": relative_path,
"includeBody": include_body,
"depth": depth,
"includeLocation": include_location,
"searchDeps": search_deps,
}
return self._make_request("POST", "/findSymbol", request_data)
def find_references(self, name_path: str, relative_path: str) -> dict[str, Any]:
"""
Finds references to a symbol.
:param name_path: the name path of the symbol
:param relative_path: the relative path
:return: dictionary containing 'symbols' list with symbol references
"""
request_data = {"namePath": name_path, "relativePath": relative_path}
return self._make_request("POST", "/findReferences", request_data)
def get_symbols_overview(self, relative_path: str, depth: int) -> dict[str, Any]:
"""
:param relative_path: the relative path to a source file
:param depth: the depth of children to include (0 = no children)
"""
request_data = {"relativePath": relative_path, "depth": depth}
return self._make_request("POST", "/getSymbolsOverview", request_data)
def rename_symbol(
self, name_path: str, relative_path: str, new_name: str, rename_in_comments: bool, rename_in_text_occurrences: bool
) -> None:
"""
Renames a symbol.
:param name_path: the name path of the symbol
:param relative_path: the relative path
:param new_name: the new name for the symbol
:param rename_in_comments: whether to rename in comments
:param rename_in_text_occurrences: whether to rename in text occurrences
"""
request_data = {
"namePath": name_path,
"relativePath": relative_path,
"newName": new_name,
"renameInComments": rename_in_comments,
"renameInTextOccurrences": rename_in_text_occurrences,
}
self._make_request("POST", "/renameSymbol", request_data)
def refresh_file(self, relative_path: str) -> None:
"""
Triggers a refresh of the given file in the IDE.
:param relative_path: the relative path
"""
request_data = {
"relativePath": relative_path,
}
self._make_request("POST", "/refreshFile", request_data)
def is_service_available(self) -> bool:
try:
self.project_root()
return True
except (ConnectionError, APIError):
return False
def close(self) -> None:
self.session.close()
def __enter__(self) -> Self:
return self
def __exit__(self, exc_type, exc_val, exc_tb): # type: ignore
self.close()
```
--------------------------------------------------------------------------------
/test/serena/test_mcp.py:
--------------------------------------------------------------------------------
```python
"""Tests for the mcp.py module in serena."""
import pytest
from mcp.server.fastmcp.tools.base import Tool as MCPTool
from serena.agent import Tool, ToolRegistry
from serena.config.context_mode import SerenaAgentContext
from serena.mcp import SerenaMCPFactory
make_tool = SerenaMCPFactory.make_mcp_tool
# Create a mock agent for tool initialization
class MockAgent:
def __init__(self):
self.project_config = None
self.serena_config = None
@staticmethod
def get_context() -> SerenaAgentContext:
return SerenaAgentContext.load_default()
class BaseMockTool(Tool):
"""A mock Tool class for testing."""
def __init__(self):
super().__init__(MockAgent()) # type: ignore
class BasicTool(BaseMockTool):
"""A mock Tool class for testing."""
def apply(self, name: str, age: int = 0) -> str:
"""This is a test function.
:param name: The person's name
:param age: The person's age
:return: A greeting message
"""
return f"Hello {name}, you are {age} years old!"
def apply_ex(
self,
log_call: bool = True,
catch_exceptions: bool = True,
**kwargs,
) -> str:
"""Mock implementation of apply_ex."""
return self.apply(**kwargs)
def test_make_tool_basic() -> None:
"""Test that make_tool correctly creates an MCP tool from a Tool object."""
mock_tool = BasicTool()
mcp_tool = make_tool(mock_tool)
# Test that the MCP tool has the correct properties
assert isinstance(mcp_tool, MCPTool)
assert mcp_tool.name == "basic"
assert "This is a test function. Returns A greeting message." in mcp_tool.description
# Test that the parameters were correctly processed
parameters = mcp_tool.parameters
assert "properties" in parameters
assert "name" in parameters["properties"]
assert "age" in parameters["properties"]
assert parameters["properties"]["name"]["description"] == "The person's name."
assert parameters["properties"]["age"]["description"] == "The person's age."
def test_make_tool_execution() -> None:
"""Test that the execution function created by make_tool works correctly."""
mock_tool = BasicTool()
mcp_tool = make_tool(mock_tool)
# Execute the MCP tool function
result = mcp_tool.fn(name="Alice", age=30)
assert result == "Hello Alice, you are 30 years old!"
def test_make_tool_no_params() -> None:
"""Test make_tool with a function that has no parameters."""
class NoParamsTool(BaseMockTool):
def apply(self) -> str:
"""This is a test function with no parameters.
:return: A simple result
"""
return "Simple result"
def apply_ex(self, *args, **kwargs) -> str:
return self.apply()
tool = NoParamsTool()
mcp_tool = make_tool(tool)
assert mcp_tool.name == "no_params"
assert "This is a test function with no parameters. Returns A simple result." in mcp_tool.description
assert mcp_tool.parameters["properties"] == {}
def test_make_tool_no_return_description() -> None:
"""Test make_tool with a function that has no return description."""
class NoReturnTool(BaseMockTool):
def apply(self, param: str) -> str:
"""This is a test function.
:param param: The parameter
"""
return f"Processed: {param}"
def apply_ex(self, *args, **kwargs) -> str:
return self.apply(**kwargs)
tool = NoReturnTool()
mcp_tool = make_tool(tool)
assert mcp_tool.name == "no_return"
assert mcp_tool.description == "This is a test function."
assert mcp_tool.parameters["properties"]["param"]["description"] == "The parameter."
def test_make_tool_parameter_not_in_docstring() -> None:
"""Test make_tool when a parameter in properties is not in the docstring."""
class MissingParamTool(BaseMockTool):
def apply(self, name: str, missing_param: str = "") -> str:
"""This is a test function.
:param name: The person's name
"""
return f"Hello {name}! Missing param: {missing_param}"
def apply_ex(self, *args, **kwargs) -> str:
return self.apply(**kwargs)
tool = MissingParamTool()
mcp_tool = make_tool(tool)
assert "name" in mcp_tool.parameters["properties"]
assert "missing_param" in mcp_tool.parameters["properties"]
assert mcp_tool.parameters["properties"]["name"]["description"] == "The person's name."
assert "description" not in mcp_tool.parameters["properties"]["missing_param"]
def test_make_tool_multiline_docstring() -> None:
"""Test make_tool with a complex multi-line docstring."""
class ComplexDocTool(BaseMockTool):
def apply(self, project_file_path: str, host: str, port: int) -> str:
"""Create an MCP server.
This function creates and configures a Model Context Protocol server
with the specified settings.
:param project_file_path: The path to the project file, or None
:param host: The host to bind to
:param port: The port to bind to
:return: A configured FastMCP server instance
"""
return f"Server config: {project_file_path}, {host}:{port}"
def apply_ex(self, *args, **kwargs) -> str:
return self.apply(**kwargs)
tool = ComplexDocTool()
mcp_tool = make_tool(tool)
assert "Create an MCP server" in mcp_tool.description
assert "Returns A configured FastMCP server instance" in mcp_tool.description
assert mcp_tool.parameters["properties"]["project_file_path"]["description"] == "The path to the project file, or None."
assert mcp_tool.parameters["properties"]["host"]["description"] == "The host to bind to."
assert mcp_tool.parameters["properties"]["port"]["description"] == "The port to bind to."
def test_make_tool_capitalization_and_periods() -> None:
"""Test that make_tool properly handles capitalization and periods in descriptions."""
class FormatTool(BaseMockTool):
def apply(self, param1: str, param2: str, param3: str) -> str:
"""Test function.
:param param1: lowercase description
:param param2: description with period.
:param param3: description with Capitalized word.
"""
return f"Formatted: {param1}, {param2}, {param3}"
def apply_ex(self, *args, **kwargs) -> str:
return self.apply(**kwargs)
tool = FormatTool()
mcp_tool = make_tool(tool)
assert mcp_tool.parameters["properties"]["param1"]["description"] == "Lowercase description."
assert mcp_tool.parameters["properties"]["param2"]["description"] == "Description with period."
assert mcp_tool.parameters["properties"]["param3"]["description"] == "Description with Capitalized word."
def test_make_tool_missing_apply() -> None:
"""Test make_tool with a tool that doesn't have an apply method."""
class BadTool(BaseMockTool):
pass
tool = BadTool()
with pytest.raises(AttributeError):
make_tool(tool)
@pytest.mark.parametrize(
"docstring, expected_description",
[
(
"""This is a test function.
:param param: The parameter
:return: A result
""",
"This is a test function. Returns A result.",
),
(
"""
:param param: The parameter
:return: A result
""",
"Returns A result.",
),
(
"""
:param param: The parameter
""",
"",
),
("Description without params.", "Description without params."),
],
)
def test_make_tool_descriptions(docstring, expected_description) -> None:
"""Test make_tool with various docstring formats."""
class TestTool(BaseMockTool):
def apply(self, param: str) -> str:
return f"Result: {param}"
def apply_ex(self, *args, **kwargs) -> str:
return self.apply(**kwargs)
# Dynamically set the docstring
TestTool.apply.__doc__ = docstring
tool = TestTool()
mcp_tool = make_tool(tool)
assert mcp_tool.name == "test"
assert mcp_tool.description == expected_description
def is_test_mock_class(tool_class: type) -> bool:
"""Check if a class is a test mock class."""
# Check if the class is defined in a test module
module_name = tool_class.__module__
return (
module_name.startswith(("test.", "tests."))
or "test_" in module_name
or tool_class.__name__
in [
"BaseMockTool",
"BasicTool",
"BadTool",
"NoParamsTool",
"NoReturnTool",
"MissingParamTool",
"ComplexDocTool",
"FormatTool",
"NoDescriptionTool",
]
)
@pytest.mark.parametrize("tool_class", ToolRegistry().get_all_tool_classes())
def test_make_tool_all_tools(tool_class) -> None:
"""Test that make_tool works for all tools in the codebase."""
# Create an instance of the tool
tool_instance = tool_class(MockAgent())
# Try to create an MCP tool from it
mcp_tool = make_tool(tool_instance)
# Basic validation
assert isinstance(mcp_tool, MCPTool)
assert mcp_tool.name == tool_class.get_name_from_cls()
# The description should be a string (either from docstring or default)
assert isinstance(mcp_tool.description, str)
```
--------------------------------------------------------------------------------
/src/solidlsp/language_servers/bash_language_server.py:
--------------------------------------------------------------------------------
```python
"""
Provides Bash specific instantiation of the LanguageServer class using bash-language-server.
Contains various configurations and settings specific to Bash scripting.
"""
import logging
import os
import pathlib
import shutil
import threading
from solidlsp.language_servers.common import RuntimeDependency, RuntimeDependencyCollection
from solidlsp.ls import DocumentSymbols, LSPFileBuffer, SolidLanguageServer
from solidlsp.ls_config import LanguageServerConfig
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
from solidlsp.settings import SolidLSPSettings
log = logging.getLogger(__name__)
class BashLanguageServer(SolidLanguageServer):
"""
Provides Bash specific instantiation of the LanguageServer class using bash-language-server.
Contains various configurations and settings specific to Bash scripting.
"""
def __init__(self, config: LanguageServerConfig, repository_root_path: str, solidlsp_settings: SolidLSPSettings):
"""
Creates a BashLanguageServer instance. This class is not meant to be instantiated directly.
Use LanguageServer.create() instead.
"""
bash_lsp_executable_path = self._setup_runtime_dependencies(config, solidlsp_settings)
super().__init__(
config,
repository_root_path,
ProcessLaunchInfo(cmd=bash_lsp_executable_path, cwd=repository_root_path),
"bash",
solidlsp_settings,
)
self.server_ready = threading.Event()
self.initialize_searcher_command_available = threading.Event()
@classmethod
def _setup_runtime_dependencies(cls, config: LanguageServerConfig, solidlsp_settings: SolidLSPSettings) -> str:
"""
Setup runtime dependencies for Bash Language Server and return the command to start the server.
"""
# Verify both node and npm are installed
is_node_installed = shutil.which("node") is not None
assert is_node_installed, "node is not installed or isn't in PATH. Please install NodeJS and try again."
is_npm_installed = shutil.which("npm") is not None
assert is_npm_installed, "npm is not installed or isn't in PATH. Please install npm and try again."
deps = RuntimeDependencyCollection(
[
RuntimeDependency(
id="bash-language-server",
description="bash-language-server package",
command="npm install --prefix ./ [email protected]",
platform_id="any",
),
]
)
# Install bash-language-server if not already installed
bash_ls_dir = os.path.join(cls.ls_resources_dir(solidlsp_settings), "bash-lsp")
bash_executable_path = os.path.join(bash_ls_dir, "node_modules", ".bin", "bash-language-server")
# Handle Windows executable extension
if os.name == "nt":
bash_executable_path += ".cmd"
if not os.path.exists(bash_executable_path):
log.info(f"Bash Language Server executable not found at {bash_executable_path}. Installing...")
deps.install(bash_ls_dir)
log.info("Bash language server dependencies installed successfully")
if not os.path.exists(bash_executable_path):
raise FileNotFoundError(
f"bash-language-server executable not found at {bash_executable_path}, something went wrong with the installation."
)
return f"{bash_executable_path} start"
@staticmethod
def _get_initialize_params(repository_absolute_path: str) -> InitializeParams:
"""
Returns the initialize params for the Bash Language Server.
"""
root_uri = pathlib.Path(repository_absolute_path).as_uri()
initialize_params = {
"locale": "en",
"capabilities": {
"textDocument": {
"synchronization": {"didSave": True, "dynamicRegistration": True},
"completion": {"dynamicRegistration": True, "completionItem": {"snippetSupport": True}},
"definition": {"dynamicRegistration": True},
"references": {"dynamicRegistration": True},
"documentSymbol": {
"dynamicRegistration": True,
"hierarchicalDocumentSymbolSupport": True,
"symbolKind": {"valueSet": list(range(1, 27))},
},
"hover": {"dynamicRegistration": True, "contentFormat": ["markdown", "plaintext"]},
"signatureHelp": {"dynamicRegistration": True},
"codeAction": {"dynamicRegistration": True},
},
"workspace": {
"workspaceFolders": True,
"didChangeConfiguration": {"dynamicRegistration": True},
"symbol": {"dynamicRegistration": True},
},
},
"processId": os.getpid(),
"rootPath": repository_absolute_path,
"rootUri": root_uri,
"workspaceFolders": [
{
"uri": root_uri,
"name": os.path.basename(repository_absolute_path),
}
],
}
return initialize_params # type: ignore
def _start_server(self) -> None:
"""
Starts the Bash Language Server, waits for the server to be ready and yields the LanguageServer instance.
"""
def register_capability_handler(params: dict) -> None:
assert "registrations" in params
for registration in params["registrations"]:
if registration["method"] == "workspace/executeCommand":
self.initialize_searcher_command_available.set()
return
def execute_client_command_handler(params: dict) -> list:
return []
def do_nothing(params: dict) -> None:
return
def window_log_message(msg: dict) -> None:
log.info(f"LSP: window/logMessage: {msg}")
# Check for bash-language-server ready signals
message_text = msg.get("message", "")
if "Analyzing" in message_text or "analysis complete" in message_text.lower():
log.info("Bash language server analysis signals detected")
self.server_ready.set()
self.completions_available.set()
self.server.on_request("client/registerCapability", register_capability_handler)
self.server.on_notification("window/logMessage", window_log_message)
self.server.on_request("workspace/executeClientCommand", execute_client_command_handler)
self.server.on_notification("$/progress", do_nothing)
self.server.on_notification("textDocument/publishDiagnostics", do_nothing)
log.info("Starting Bash server process")
self.server.start()
initialize_params = self._get_initialize_params(self.repository_root_path)
log.info("Sending initialize request from LSP client to LSP server and awaiting response")
init_response = self.server.send.initialize(initialize_params)
log.debug(f"Received initialize response from bash server: {init_response}")
# Enhanced capability checks for bash-language-server 5.6.0
assert init_response["capabilities"]["textDocumentSync"] in [1, 2] # Full or Incremental
assert "completionProvider" in init_response["capabilities"]
# Verify document symbol support is available
if "documentSymbolProvider" in init_response["capabilities"]:
log.info("Bash server supports document symbols")
else:
log.warning("Warning: Bash server does not report document symbol support")
self.server.notify.initialized({})
# Wait for server readiness with timeout
log.info("Waiting for Bash language server to be ready...")
if not self.server_ready.wait(timeout=3.0):
# Fallback: assume server is ready after timeout
# This is common. bash-language-server doesn't always send explicit ready signals. Log as info
log.info("Timeout waiting for bash server ready signal, proceeding anyway")
self.server_ready.set()
self.completions_available.set()
else:
log.info("Bash server initialization complete")
def request_document_symbols(self, relative_file_path: str, file_buffer: LSPFileBuffer | None = None) -> DocumentSymbols:
# Uses the standard LSP documentSymbol request which provides reliable function detection
# for all bash function syntaxes including:
# - function name() { ... } (with function keyword)
# - name() { ... } (traditional syntax)
# - Functions with various indentation levels
# - Functions with comments before/after/inside
log.debug(f"Requesting document symbols via LSP for {relative_file_path}")
# Use the standard LSP approach - bash-language-server handles all function syntaxes correctly
document_symbols = super().request_document_symbols(relative_file_path, file_buffer=file_buffer)
# Log detection results for debugging
functions = [s for s in document_symbols.iter_symbols() if s.get("kind") == 12]
log.info(f"LSP function detection for {relative_file_path}: Found {len(functions)} functions")
return document_symbols
```
--------------------------------------------------------------------------------
/src/serena/task_executor.py:
--------------------------------------------------------------------------------
```python
import concurrent.futures
import threading
import time
from collections.abc import Callable
from concurrent.futures import Future
from dataclasses import dataclass
from threading import Thread
from typing import Generic, TypeVar
from sensai.util import logging
from sensai.util.logging import LogTime
from sensai.util.string import ToStringMixin
log = logging.getLogger(__name__)
T = TypeVar("T")
class TaskExecutor:
def __init__(self, name: str):
self._task_executor_lock = threading.Lock()
self._task_executor_queue: list[TaskExecutor.Task] = []
self._task_executor_thread = Thread(target=self._process_task_queue, name=name, daemon=True)
self._task_executor_thread.start()
self._task_executor_task_index = 1
self._task_executor_current_task: TaskExecutor.Task | None = None
self._task_executor_last_executed_task_info: TaskExecutor.TaskInfo | None = None
class Task(ToStringMixin, Generic[T]):
def __init__(self, function: Callable[[], T], name: str, logged: bool = True, timeout: float | None = None):
"""
:param function: the function representing the task to execute
:param name: the name of the task
:param logged: whether to log management of the task; if False, only errors will be logged
:param timeout: the maximum time to wait for task completion in seconds, or None to wait indefinitely
"""
self.name = name
self.future: concurrent.futures.Future = concurrent.futures.Future()
self.logged = logged
self.timeout = timeout
self._function = function
def _tostring_includes(self) -> list[str]:
return ["name"]
def start(self) -> None:
"""
Executes the task in a separate thread, setting the result or exception on the future.
"""
def run_task() -> None:
try:
if self.future.done():
if self.logged:
log.info(f"Task {self.name} was already completed/cancelled; skipping execution")
return
with LogTime(self.name, logger=log, enabled=self.logged):
result = self._function()
if not self.future.done():
self.future.set_result(result)
except Exception as e:
if not self.future.done():
log.error(f"Error during execution of {self.name}: {e}", exc_info=e)
self.future.set_exception(e)
thread = Thread(target=run_task, name=self.name)
thread.start()
def is_done(self) -> bool:
"""
:return: whether the task has completed (either successfully, with failure, or via cancellation)
"""
return self.future.done()
def result(self, timeout: float | None = None) -> T:
"""
Blocks until the task is done or the timeout is reached, and returns the result.
If an exception occurred during task execution, it is raised here.
If the timeout is reached, a TimeoutError is raised (but the task is not cancelled).
If the task is cancelled, a CancelledError is raised.
:param timeout: the maximum time to wait in seconds; if None, use the task's own timeout
(which may be None to wait indefinitely)
:return: True if the task is done, False if the timeout was reached
"""
return self.future.result(timeout=timeout)
def cancel(self) -> None:
"""
Cancels the task. If it has not yet started, it will not be executed.
If it has already started, its future will be marked as cancelled and will raise a CancelledError
when its result is requested.
"""
self.future.cancel()
def wait_until_done(self, timeout: float | None = None) -> None:
"""
Waits until the task is done or the timeout is reached.
The task is done if it either completed successfully, failed with an exception, or was cancelled.
:param timeout: the maximum time to wait in seconds; if None, use the task's own timeout
(which may be None to wait indefinitely)
"""
try:
self.future.result(timeout=timeout)
except:
pass
def _process_task_queue(self) -> None:
while True:
# obtain task from the queue
task: TaskExecutor.Task | None = None
with self._task_executor_lock:
if len(self._task_executor_queue) > 0:
task = self._task_executor_queue.pop(0)
if task is None:
time.sleep(0.1)
continue
# start task execution asynchronously
with self._task_executor_lock:
self._task_executor_current_task = task
if task.logged:
log.info("Starting execution of %s", task.name)
task.start()
# wait for task completion
task.wait_until_done(timeout=task.timeout)
with self._task_executor_lock:
self._task_executor_current_task = None
if task.logged:
self._task_executor_last_executed_task_info = self.TaskInfo.from_task(task, is_running=False)
@dataclass
class TaskInfo:
name: str
is_running: bool
future: Future
"""
future for accessing the task's result
"""
task_id: int
"""
unique identifier of the task
"""
logged: bool
def finished_successfully(self) -> bool:
return self.future.done() and not self.future.cancelled() and self.future.exception() is None
@staticmethod
def from_task(task: "TaskExecutor.Task", is_running: bool) -> "TaskExecutor.TaskInfo":
return TaskExecutor.TaskInfo(name=task.name, is_running=is_running, future=task.future, task_id=id(task), logged=task.logged)
def cancel(self) -> None:
self.future.cancel()
def get_current_tasks(self) -> list[TaskInfo]:
"""
Gets the list of tasks currently running or queued for execution.
The function returns a list of thread-safe TaskInfo objects (specifically created for the caller).
:return: the list of tasks in the execution order (running task first)
"""
tasks = []
with self._task_executor_lock:
if self._task_executor_current_task is not None:
tasks.append(self.TaskInfo.from_task(self._task_executor_current_task, True))
for task in self._task_executor_queue:
if not task.is_done():
tasks.append(self.TaskInfo.from_task(task, False))
return tasks
def issue_task(self, task: Callable[[], T], name: str | None = None, logged: bool = True, timeout: float | None = None) -> Task[T]:
"""
Issue a task to the executor for asynchronous execution.
It is ensured that tasks are executed in the order they are issued, one after another.
:param task: the task to execute
:param name: the name of the task for logging purposes; if None, use the task function's name
:param logged: whether to log management of the task; if False, only errors will be logged
:param timeout: the maximum time to wait for task completion in seconds, or None to wait indefinitely
:return: the task object, through which the task's future result can be accessed
"""
with self._task_executor_lock:
if logged:
task_prefix_name = f"Task-{self._task_executor_task_index}"
self._task_executor_task_index += 1
else:
task_prefix_name = "BackgroundTask"
task_name = f"{task_prefix_name}:{name or task.__name__}"
if logged:
log.info(f"Scheduling {task_name}")
task_obj = self.Task(function=task, name=task_name, logged=logged, timeout=timeout)
self._task_executor_queue.append(task_obj)
return task_obj
def execute_task(self, task: Callable[[], T], name: str | None = None, logged: bool = True, timeout: float | None = None) -> T:
"""
Executes the given task synchronously via the agent's task executor.
This is useful for tasks that need to be executed immediately and whose results are needed right away.
:param task: the task to execute
:param name: the name of the task for logging purposes; if None, use the task function's name
:param logged: whether to log management of the task; if False, only errors will be logged
:param timeout: the maximum time to wait for task completion in seconds, or None to wait indefinitely
:return: the result of the task execution
"""
task_obj = self.issue_task(task, name=name, logged=logged, timeout=timeout)
return task_obj.result()
def get_last_executed_task(self) -> TaskInfo | None:
"""
Gets information about the last executed task.
:return: TaskInfo of the last executed task, or None if no task has been executed yet.
"""
with self._task_executor_lock:
return self._task_executor_last_executed_task_info
```