hermes-agent/agent/lsp/servers.py
Teknium 83b93898c2
feat(lsp): semantic diagnostics from real language servers in write_file/patch (#24168)
* feat(lsp): semantic diagnostics from real language servers in write_file/patch

Wire ~26 language servers (pyright, gopls, rust-analyzer, typescript-language-server,
clangd, bash-language-server, ...) into the post-write lint check used by write_file
and patch. The model now sees type errors, undefined names, missing imports, and
project-wide semantic issues introduced by its edits, not just syntax errors.

LSP is gated on git workspace detection: when the agent's cwd or the file being
edited is inside a git worktree, LSP runs against that workspace; otherwise the
existing in-process syntax checks are the only tier. This keeps users on
user-home cwds (Telegram/Discord gateway chats) from spawning daemons.

The post-write check is layered: in-process syntax check first (microseconds),
then LSP semantic diagnostics second when syntax is clean. Diagnostics are
delta-filtered against a baseline captured at write start, so the agent only
sees errors its edit introduced. A flaky/missing language server can never
break a write -- every LSP failure path falls back silently to the syntax-only
result.

New module agent/lsp/ split into:

- protocol.py: Content-Length JSON-RPC framer + envelope helpers
- client.py: async LSPClient (spawn, initialize, didOpen/didChange,
  ContentModified retry, push/pull diagnostic stores)
- workspace.py: git worktree walk-up + per-server NearestRoot resolver
- servers.py: registry of 26 language servers (extension match,
  root resolver, spawn builder per language)
- install.py: auto-install dispatch (npm install --prefix, go install
  with GOBIN, pip install --target) into HERMES_HOME/lsp/bin/
- manager.py: LSPService (per-(server_id, root) client registry, lazy
  spawn, broken-set, in-flight dedupe, sync facade for tools layer)
- reporter.py: <diagnostics> block formatter (severity-1-only, 20-per-file)
- cli.py: hermes lsp {status,list,install,install-all,restart,which}

Wired into tools/file_operations.py:

- write_file/patch_replace now call _snapshot_lsp_baseline before write
- _check_lint_delta gains a third tier: LSP semantic diagnostics when
  syntax is clean
- All LSP code paths swallow exceptions; write_file's contract unchanged

Config: 'lsp' section in DEFAULT_CONFIG with enabled (default true),
wait_mode, wait_timeout, install_strategy (default 'auto'), and per-server
overrides (disabled, command, env, initialization_options).

Tests: tests/agent/lsp/ -- 49 tests covering protocol framing (encode and
read_message round-trip, EOF/truncation/missing Content-Length), workspace
gate (git walk-up, exclude markers, fallback to file location), reporter
(severity filter, max-per-file cap, truncation), service-level delta filter,
and an in-process mock LSP server that exercises the full client lifecycle
including didChange version bumps, dedup, crash recovery, and idempotent
teardown.

Live E2E verified end-to-end through ShellFileOperations: pyright
auto-installed via npm into HERMES_HOME, baseline captured, type error
introduced, single delta diagnostic surfaced with correct line/column/code/
source, then patch fix removes the diagnostic from the output.

Docs: new website/docs/user-guide/features/lsp.md page covering supported
languages, configuration knobs, performance characteristics, and
troubleshooting; cli-commands.md updated with the 'hermes lsp' reference;
sidebar updated.

* feat(lsp): structured logging, backend gate, defensive walk caps

Cherry-picks the substantive ideas from #24155 (different scope, same
problem space) onto our PR.

agent/lsp/eventlog.py (new): dedicated structured logger
``hermes.lint.lsp`` with steady-state silence. Module-level dedup sets
keep a 1000-write session at exactly ONE INFO line ("active for
<root>") at the default INFO threshold; clean writes log at DEBUG so
they never reach agent.log under normal config. State transitions
(server starts, no project root for a file, server unavailable) fire
at INFO/WARNING once per (server_id, key); novel events (timeouts,
unexpected errors) fire WARNING per call. Grep recipe: ``rg 'lsp\\['``.

agent/lsp/manager.py: wire the eventlog into _get_or_spawn and
get_diagnostics_sync so users can answer "did LSP fire on this edit?"
with a single grep, plus surface "binary not on PATH" warnings once
instead of silently retrying every write.

tools/file_operations.py: backend-type gate. ``_lsp_local_only()``
returns False for non-local backends (Docker / Modal / SSH /
Daytona); ``_snapshot_lsp_baseline`` and ``_maybe_lsp_diagnostics``
now skip entirely on remote envs. The host-side language server
can't see files inside a sandbox, so this prevents pretending to
lint a file the host process can't open.

agent/lsp/protocol.py: 8 KiB cap on the header block in
``read_message``. A pathological server that streams headers
without ever emitting CRLF-CRLF would have looped forever consuming
bytes; now raises ``LSPProtocolError`` instead.

agent/lsp/workspace.py: 64-step cap on ``find_git_worktree`` and
``nearest_root`` upward walks, plus try/except containment around
``Path(...).resolve()`` and child ``.exists()`` calls. Defensive
against pathological inputs (symlink loops, encoding errors,
permission failures mid-walk) — the lint hook is hot-path code and
must never raise.

Tests:
- tests/agent/lsp/test_eventlog.py: 18 tests covering steady-state
  silence (clean writes stay DEBUG), state-transition INFO-once
  semantics (active for, no project root), action-required
  WARNING-once (server unavailable), per-call WARNING (timeouts,
  spawn failures), and the "1000 clean writes => 1 INFO" contract.
- tests/agent/lsp/test_backend_gate.py: 5 tests verifying
  _lsp_local_only / snapshot_baseline / maybe_lsp_diagnostics skip
  the LSP layer for non-local backends and route correctly for
  LocalEnvironment.
- tests/agent/lsp/test_protocol.py: new test_read_message_rejects_runaway_header
  exercising the 8 KiB cap.

Validation:
- 73/73 LSP tests pass (49 original + 18 eventlog + 5 backend-gate + 1 framer cap)
- 198/198 pass when run alongside existing file_operations tests
- Live E2E re-run with pyright still surfaces "ERROR [2:12] Type
  ... reportReturnType (Pyright)" through the full path, then patch
  fix removes it on the next call.

* feat(lsp): atexit cleanup + separate lsp_diagnostics JSON field

Two improvements salvaged from #24414's plugin-form alternative,
keeping our core-integrated design:

1. atexit cleanup of spawned language servers
   ----------------------------------------------------------------
   ``agent/lsp/__init__.get_service`` now registers an ``atexit``
   handler on first creation that tears down the LSPService on
   Python exit.  Without this, every ``hermes chat`` exit was
   leaking pyright/gopls/etc. processes for a few seconds while
   their stdout buffers drained -- they got reaped by the kernel
   eventually but a watchful ``ps aux`` would catch them.

   The handler runs once per process (gated by
   ``_atexit_registered``); idempotent ``shutdown_service``
   ensures double-fire is a no-op.  Errors during shutdown are
   swallowed at debug level since by the time atexit fires the
   user has already seen the agent's final response.

2. Separate ``lsp_diagnostics`` field on WriteResult / PatchResult
   ----------------------------------------------------------------
   Previously the LSP layer folded its diagnostic block into the
   ``lint.output`` string, conflating the syntax-check tier with
   the semantic tier.  The agent (and any downstream parsers) now
   read syntax errors and semantic errors as independent signals:

       {
         "bytes_written": 42,
         "lint": {"status": "ok", "output": ""},
         "lsp_diagnostics": "<diagnostics file=...>\nERROR [2:12] ..."
       }

   ``_check_lint_delta`` returns to its original two-tier shape
   (syntax check + delta filter); ``write_file`` and
   ``patch_replace`` independently fetch LSP diagnostics via
   ``_maybe_lsp_diagnostics`` and pass them into the new field.
   ``patch_replace`` propagates the inner write_file's
   ``lsp_diagnostics`` so the outer PatchResult carries the patch's
   delta correctly.

Tests: 19 new
- tests/agent/lsp/test_lifecycle.py (8 tests): atexit registration
  fires once and only once across N get_service calls; the
  registered callable is our internal shutdown wrapper;
  shutdown_service is idempotent and safe when never started;
  exceptions during shutdown are swallowed; inactive service is
  cached so we don't rebuild on every check.
- tests/agent/lsp/test_diagnostics_field.py (11 tests): WriteResult
  / PatchResult dataclass shape, to_dict include/omit semantics,
  channel separation (lint and lsp_diagnostics carry independent
  signals), write_file populates the field via
  _maybe_lsp_diagnostics only when the syntax tier is clean,
  patch_replace propagates the field forward from its internal
  write_file.

Validation:
- 92/92 LSP tests pass (73 prior + 8 lifecycle + 11 diagnostics field)
- 217/217 pass with file_operations + LSP combined
- Live E2E reverified: clean writes -> both fields empty/none; type
  error introduced -> lint clean (parses), lsp_diagnostics carries
  the pyright reportReturnType block; patch fix -> both fields
  clean again.

* fix(lsp): broken-set short-circuit so a wedged server isn't paid every write

Discovered while auditing failure paths: a language server binary that
hangs (sleep forever, no LSP traffic on stdin/stdout) caused EVERY
subsequent write to re-pay the 8s snapshot_baseline timeout. Five
writes = ~64s of dead time.

The bug: ``_get_or_spawn`` adds the (server_id, root) pair to
``_broken`` inside its inner exception handler, but when the OUTER
``_loop.run`` timeout fires, it cancels the inner task before that
handler runs. The pair never makes it to broken-set, so the next
write re-enters the spawn path and re-pays the timeout.

Fix:

- New ``_mark_broken_for_file`` helper at the service layer marks
  the (server_id, workspace_root) pair broken from the OUTSIDE when
  the outer timeout fires. Called from the except branches in
  ``snapshot_baseline``, ``get_diagnostics_sync`` (asyncio.TimeoutError
  + generic Exception). Also kills any orphan client process that
  survived the cancelled future, fire-and-forget with a 1s ceiling.

- ``enabled_for`` now consults the broken-set BEFORE returning True.
  Files in already-broken (server_id, root) pairs short-circuit to
  False, so the file_operations layer skips the LSP path entirely
  with no spawn cost. Until the service is restarted (``hermes lsp
  restart``) or the process exits.

- A single eventlog WARNING is emitted on first mark-broken so the
  user knows which server gave up. Subsequent edits in the same
  project stay silent.

Tests: 7 new in tests/agent/lsp/test_broken_set.py — covers the
key shape (server_id, per_server_root), enabled_for short-circuit,
sibling-file skip in same project, project isolation (broken in
A doesn't affect B), graceful no-op for missing-server / no-workspace,
and an end-to-end test that snapshots after a failure and verifies
the next ``enabled_for`` returns False.

Validation:

- Live retest of the wedged-binary scenario: 5 sequential writes,
  first 8.88s (the one snapshot timeout), subsequent four ~0.84s
  (no LSP cost). Down from 5x12.85s = 64s before this fix.
- 99/99 LSP tests pass (92 prior + 7 broken-set)
- 224/224 pass with file_operations + LSP combined
- Happy path E2E reverified — clean write, type error introduced,
  patch fix all behave correctly with the new broken-set logic.

Note: the FIRST write to a wedged binary still pays 8s (the
snapshot_baseline timeout). We could shorten that, but pyright/
tsserver normally take 2-3s and slow CI rust-analyzer can need
5+ seconds, so 8s is the conservative ceiling. Subsequent writes
are instant.
2026-05-12 16:31:54 -07:00

1025 lines
33 KiB
Python

"""Server registry — per-language LSP server definitions.
Each :class:`ServerDef` knows how to:
- match a file by extension (or basename for extensionless files like
``Dockerfile``),
- resolve a project root from a file path (often via
:func:`agent.lsp.workspace.nearest_root`),
- assemble the spawn command (binary, args, env, cwd),
- compute LSP ``initializationOptions``.
Auto-installation is a separate concern handled by
:mod:`agent.lsp.install`. This module describes WHAT to spawn; the
install module makes the binary appear on PATH if it isn't there.
The full set of servers ships with the package, but most are only
*invoked* when the user actually edits a file in that language. This
keeps cold-start fast — we don't probe binaries until needed.
"""
from __future__ import annotations
import logging
import os
import shutil
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple
from agent.lsp.workspace import nearest_root, normalize_path
logger = logging.getLogger("agent.lsp.servers")
# Language IDs per LSP spec. Used for ``textDocument/didOpen.languageId``.
# Most servers don't care exactly, but a few (typescript-language-server,
# vue-language-server) refuse files with the wrong ID.
LANGUAGE_BY_EXT: Dict[str, str] = {
".py": "python",
".pyi": "python",
".ts": "typescript",
".tsx": "typescriptreact",
".js": "javascript",
".jsx": "javascriptreact",
".mjs": "javascript",
".cjs": "javascript",
".mts": "typescript",
".cts": "typescript",
".vue": "vue",
".svelte": "svelte",
".astro": "astro",
".go": "go",
".rs": "rust",
".rb": "ruby",
".rake": "ruby",
".gemspec": "ruby",
".ru": "ruby",
".c": "c",
".h": "c",
".cc": "cpp",
".cpp": "cpp",
".cxx": "cpp",
".hh": "cpp",
".hpp": "cpp",
".hxx": "cpp",
".cs": "csharp",
".csx": "csharp",
".fs": "fsharp",
".fsi": "fsharp",
".fsx": "fsharp",
".swift": "swift",
".java": "java",
".kt": "kotlin",
".kts": "kotlin",
".yaml": "yaml",
".yml": "yaml",
".json": "json",
".jsonc": "jsonc",
".lua": "lua",
".php": "php",
".prisma": "prisma",
".dart": "dart",
".ml": "ocaml",
".mli": "ocaml",
".sh": "shellscript",
".bash": "shellscript",
".zsh": "shellscript",
".tf": "terraform",
".tfvars": "terraform",
".tex": "latex",
".bib": "bibtex",
".gleam": "gleam",
".clj": "clojure",
".cljs": "clojurescript",
".cljc": "clojure",
".edn": "clojure",
".nix": "nix",
".typ": "typst",
".typc": "typst",
".hs": "haskell",
".lhs": "haskell",
".jl": "julia",
".ex": "elixir",
".exs": "elixir",
".zig": "zig",
".zon": "zig",
".dockerfile": "dockerfile",
}
@dataclass
class SpawnSpec:
"""The result of resolving a server for a file.
Returned by :meth:`ServerDef.resolve` when a server is applicable
to a file. ``None`` is returned instead when the server should
be skipped (binary missing and auto-install disabled, project
marker not found, exclude marker hit, etc.).
"""
command: List[str]
workspace_root: str
cwd: str
env: Dict[str, str] = field(default_factory=dict)
initialization_options: Dict[str, Any] = field(default_factory=dict)
seed_diagnostics_on_first_push: bool = False
@dataclass
class ServerDef:
"""Definition of one language server.
The :func:`resolve_root` callable receives the absolute file path
plus the workspace root (git worktree) and returns either the
project-specific root for this server (e.g. the directory
containing ``pyproject.toml``) or ``None`` to skip.
The :func:`build_spawn` callable receives the resolved root and
returns a :class:`SpawnSpec` (or ``None`` if the binary can't be
found and auto-install isn't configured).
"""
server_id: str
extensions: Tuple[str, ...]
resolve_root: Callable[[str, str], Optional[str]]
build_spawn: Callable[[str, "ServerContext"], Optional[SpawnSpec]]
seed_first_push: bool = False
description: str = ""
def matches(self, file_path: str) -> bool:
"""Return True iff this server handles ``file_path``."""
ext = _file_ext_or_basename(file_path)
return ext in self.extensions
@dataclass
class ServerContext:
"""Context passed into :meth:`ServerDef.build_spawn`.
Carries the user's auto-install policy, any user-overridden
binary paths, and helpers the spawn builder needs. All fields
are optional; defaults yield "auto-install allowed, no overrides".
"""
workspace_root: str
install_strategy: str = "auto" # "auto" | "manual" | "off"
binary_overrides: Dict[str, List[str]] = field(default_factory=dict)
env_overrides: Dict[str, Dict[str, str]] = field(default_factory=dict)
init_overrides: Dict[str, Dict[str, Any]] = field(default_factory=dict)
# ---------------------------------------------------------------------------
# helpers
# ---------------------------------------------------------------------------
def _file_ext_or_basename(path: str) -> str:
"""Return the lower-cased extension OR full basename for extensionless files.
Mirrors OpenCode's ``path.parse(file).ext || file`` — files like
``Dockerfile`` or ``Makefile`` match by basename, while normal
files match by extension (``.py``, ``.ts``).
"""
base = os.path.basename(path)
_root, ext = os.path.splitext(base)
if ext:
return ext.lower()
return base
def _which(*names: str) -> Optional[str]:
"""Return the full path of the first command found on PATH."""
for n in names:
path = shutil.which(n)
if path:
return path
return None
def _root_or_workspace(file_path: str, workspace: str, markers: Sequence[str], excludes: Sequence[str] = ()) -> Optional[str]:
"""Common pattern: try ``nearest_root``, fall back to workspace root.
Returns ``None`` if an exclude marker matches first (server gated off).
"""
found = nearest_root(
file_path,
markers,
excludes=excludes,
ceiling=os.path.dirname(workspace) if workspace else None,
)
if found is None and excludes:
# Distinguish "no marker found" from "exclude hit": when
# excludes are configured, None means gated off.
# Re-check without excludes — if still None, we fall back to
# workspace; if found, the exclude hit and we return None.
recheck = nearest_root(
file_path,
markers,
ceiling=os.path.dirname(workspace) if workspace else None,
)
if recheck is not None:
return None # exclude triggered
return workspace
return found or workspace
# ---------------------------------------------------------------------------
# per-server spawn builders
# ---------------------------------------------------------------------------
def _spawn_pyright(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "pyright") or _which(
"pyright-langserver", "pyright"
)
if bin_path is None:
from agent.lsp.install import try_install
bin_path = try_install("pyright", ctx.install_strategy)
if bin_path is None:
return None
# If we got the cli ``pyright``, the langserver is its sibling.
base = os.path.basename(bin_path)
if base in ("pyright", "pyright.exe"):
sibling = os.path.join(os.path.dirname(bin_path), "pyright-langserver")
if os.path.exists(sibling):
bin_path = sibling
init: Dict[str, Any] = {}
# Pick the project's venv interpreter if there is one — otherwise
# pyright defaults to "python on PATH" which is rarely the venv.
py = _detect_python(root)
if py:
init["python"] = {"pythonPath": py}
if "pyright" in ctx.init_overrides:
init.update(ctx.init_overrides["pyright"])
return SpawnSpec(
command=[bin_path, "--stdio"],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("pyright", {}),
initialization_options=init,
)
def _detect_python(root: str) -> Optional[str]:
candidates = []
if os.environ.get("VIRTUAL_ENV"):
candidates.append(os.environ["VIRTUAL_ENV"])
candidates.extend([os.path.join(root, ".venv"), os.path.join(root, "venv")])
for v in candidates:
for sub in ("bin/python", "bin/python3", "Scripts/python.exe"):
p = os.path.join(v, sub)
if os.path.exists(p):
return p
return None
def _spawn_typescript(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "typescript") or _which("typescript-language-server")
if bin_path is None:
from agent.lsp.install import try_install
bin_path = try_install("typescript-language-server", ctx.install_strategy)
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path, "--stdio"],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("typescript", {}),
initialization_options=ctx.init_overrides.get("typescript", {}),
seed_diagnostics_on_first_push=True,
)
def _spawn_gopls(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "gopls") or _which("gopls")
if bin_path is None:
from agent.lsp.install import try_install
bin_path = try_install("gopls", ctx.install_strategy)
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("gopls", {}),
initialization_options=ctx.init_overrides.get("gopls", {}),
)
def _spawn_rust_analyzer(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "rust-analyzer") or _which("rust-analyzer")
if bin_path is None:
from agent.lsp.install import try_install
bin_path = try_install("rust-analyzer", ctx.install_strategy)
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("rust-analyzer", {}),
initialization_options=ctx.init_overrides.get("rust-analyzer", {}),
)
def _spawn_clangd(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "clangd") or _which("clangd")
if bin_path is None:
from agent.lsp.install import try_install
bin_path = try_install("clangd", ctx.install_strategy)
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path, "--background-index", "--clang-tidy"],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("clangd", {}),
initialization_options=ctx.init_overrides.get("clangd", {}),
)
def _spawn_bash_ls(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "bash-language-server") or _which("bash-language-server")
if bin_path is None:
from agent.lsp.install import try_install
bin_path = try_install("bash-language-server", ctx.install_strategy)
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path, "start"],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("bash-language-server", {}),
initialization_options=ctx.init_overrides.get("bash-language-server", {}),
)
def _spawn_yaml_ls(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "yaml-language-server") or _which("yaml-language-server")
if bin_path is None:
from agent.lsp.install import try_install
bin_path = try_install("yaml-language-server", ctx.install_strategy)
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path, "--stdio"],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("yaml-language-server", {}),
initialization_options=ctx.init_overrides.get("yaml-language-server", {}),
)
def _spawn_lua_ls(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "lua-language-server") or _which("lua-language-server")
if bin_path is None:
from agent.lsp.install import try_install
bin_path = try_install("lua-language-server", ctx.install_strategy)
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("lua-language-server", {}),
initialization_options=ctx.init_overrides.get("lua-language-server", {}),
)
def _spawn_intelephense(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "intelephense") or _which("intelephense")
if bin_path is None:
from agent.lsp.install import try_install
bin_path = try_install("intelephense", ctx.install_strategy)
if bin_path is None:
return None
init = {"telemetry": {"enabled": False}}
init.update(ctx.init_overrides.get("intelephense", {}))
return SpawnSpec(
command=[bin_path, "--stdio"],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("intelephense", {}),
initialization_options=init,
)
def _spawn_ocamllsp(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "ocaml-lsp") or _which("ocamllsp")
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("ocaml-lsp", {}),
initialization_options=ctx.init_overrides.get("ocaml-lsp", {}),
)
def _spawn_dockerfile_ls(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "dockerfile-ls") or _which("docker-langserver")
if bin_path is None:
from agent.lsp.install import try_install
bin_path = try_install("dockerfile-language-server-nodejs", ctx.install_strategy)
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path, "--stdio"],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("dockerfile-ls", {}),
initialization_options=ctx.init_overrides.get("dockerfile-ls", {}),
)
def _spawn_terraform_ls(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "terraform-ls") or _which("terraform-ls")
if bin_path is None:
return None # terraform-ls is heavy to auto-install; require user
init = {
"experimentalFeatures": {
"prefillRequiredFields": True,
"validateOnSave": True,
}
}
init.update(ctx.init_overrides.get("terraform-ls", {}))
return SpawnSpec(
command=[bin_path, "serve"],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("terraform-ls", {}),
initialization_options=init,
)
def _spawn_dart(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "dart") or _which("dart")
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path, "language-server", "--lsp"],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("dart", {}),
initialization_options=ctx.init_overrides.get("dart", {}),
)
def _spawn_haskell_ls(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "haskell-language-server") or _which(
"haskell-language-server-wrapper", "haskell-language-server"
)
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path, "--lsp"],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("haskell-language-server", {}),
initialization_options=ctx.init_overrides.get("haskell-language-server", {}),
)
def _spawn_julia(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "julia") or _which("julia")
if bin_path is None:
return None
return SpawnSpec(
command=[
bin_path,
"--startup-file=no",
"--history-file=no",
"-e",
"using LanguageServer; runserver()",
],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("julia", {}),
initialization_options=ctx.init_overrides.get("julia", {}),
)
def _spawn_clojure_lsp(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "clojure-lsp") or _which("clojure-lsp")
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path, "listen"],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("clojure-lsp", {}),
initialization_options=ctx.init_overrides.get("clojure-lsp", {}),
)
def _spawn_nixd(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "nixd") or _which("nixd")
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("nixd", {}),
initialization_options=ctx.init_overrides.get("nixd", {}),
)
def _spawn_zls(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "zls") or _which("zls")
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("zls", {}),
initialization_options=ctx.init_overrides.get("zls", {}),
)
def _spawn_gleam(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "gleam") or _which("gleam")
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path, "lsp"],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("gleam", {}),
initialization_options=ctx.init_overrides.get("gleam", {}),
)
def _spawn_elixir_ls(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "elixir-ls") or _which("elixir-ls", "language_server.sh")
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("elixir-ls", {}),
initialization_options=ctx.init_overrides.get("elixir-ls", {}),
)
def _spawn_prisma(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "prisma") or _which("prisma")
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path, "language-server"],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("prisma", {}),
initialization_options=ctx.init_overrides.get("prisma", {}),
)
def _spawn_kotlin_ls(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "kotlin-language-server") or _which(
"kotlin-language-server"
)
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("kotlin-language-server", {}),
initialization_options=ctx.init_overrides.get("kotlin-language-server", {}),
)
def _spawn_jdtls(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
# jdtls has a complex install flow. We require a manual install
# for now and look for the wrapper script that the jdtls install
# produces.
bin_path = _resolve_override(ctx, "jdtls") or _which("jdtls")
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("jdtls", {}),
initialization_options=ctx.init_overrides.get("jdtls", {}),
)
def _spawn_vue(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "vue-language-server") or _which(
"vue-language-server"
)
if bin_path is None:
from agent.lsp.install import try_install
bin_path = try_install("@vue/language-server", ctx.install_strategy)
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path, "--stdio"],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("vue-language-server", {}),
initialization_options=ctx.init_overrides.get("vue-language-server", {}),
)
def _spawn_svelte(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "svelte-language-server") or _which(
"svelteserver", "svelte-language-server"
)
if bin_path is None:
from agent.lsp.install import try_install
bin_path = try_install("svelte-language-server", ctx.install_strategy)
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path, "--stdio"],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("svelte-language-server", {}),
initialization_options=ctx.init_overrides.get("svelte-language-server", {}),
)
def _spawn_astro(root: str, ctx: ServerContext) -> Optional[SpawnSpec]:
bin_path = _resolve_override(ctx, "astro-language-server") or _which(
"astro-ls", "astro-language-server"
)
if bin_path is None:
from agent.lsp.install import try_install
bin_path = try_install("@astrojs/language-server", ctx.install_strategy)
if bin_path is None:
return None
return SpawnSpec(
command=[bin_path, "--stdio"],
workspace_root=root,
cwd=root,
env=ctx.env_overrides.get("astro-language-server", {}),
initialization_options=ctx.init_overrides.get("astro-language-server", {}),
)
def _resolve_override(ctx: ServerContext, server_id: str) -> Optional[str]:
"""User can pin a binary path in config."""
override = ctx.binary_overrides.get(server_id)
if override and override[0] and os.path.exists(override[0]):
return override[0]
return None
# ---------------------------------------------------------------------------
# root resolvers
# ---------------------------------------------------------------------------
def _root_python(file_path: str, workspace: str) -> Optional[str]:
return _root_or_workspace(
file_path,
workspace,
["pyproject.toml", "setup.py", "setup.cfg", "requirements.txt", "Pipfile", "pyrightconfig.json"],
)
def _root_typescript(file_path: str, workspace: str) -> Optional[str]:
return _root_or_workspace(
file_path,
workspace,
[
"package-lock.json",
"bun.lockb",
"bun.lock",
"pnpm-lock.yaml",
"yarn.lock",
"package.json",
"tsconfig.json",
],
excludes=["deno.json", "deno.jsonc"],
)
def _root_go(file_path: str, workspace: str) -> Optional[str]:
return _root_or_workspace(
file_path,
workspace,
["go.work", "go.mod", "go.sum"],
)
def _root_rust(file_path: str, workspace: str) -> Optional[str]:
return _root_or_workspace(file_path, workspace, ["Cargo.toml", "Cargo.lock"])
def _root_ruby(file_path: str, workspace: str) -> Optional[str]:
return _root_or_workspace(file_path, workspace, ["Gemfile"])
def _root_clangd(file_path: str, workspace: str) -> Optional[str]:
return _root_or_workspace(
file_path,
workspace,
["compile_commands.json", "compile_flags.txt", ".clangd"],
)
def _root_bash(file_path: str, workspace: str) -> str:
return workspace
def _root_yaml(file_path: str, workspace: str) -> str:
return workspace
def _root_lua(file_path: str, workspace: str) -> Optional[str]:
return _root_or_workspace(
file_path,
workspace,
[".luarc.json", ".luarc.jsonc", ".luacheckrc", ".stylua.toml", "stylua.toml", "selene.toml", "selene.yml"],
)
def _root_php(file_path: str, workspace: str) -> Optional[str]:
return _root_or_workspace(file_path, workspace, ["composer.json", "composer.lock", ".php-version"])
def _root_ocaml(file_path: str, workspace: str) -> Optional[str]:
return _root_or_workspace(file_path, workspace, ["dune-project", "dune-workspace", ".merlin", "opam"])
def _root_docker(file_path: str, workspace: str) -> str:
return workspace
def _root_terraform(file_path: str, workspace: str) -> Optional[str]:
return _root_or_workspace(file_path, workspace, [".terraform.lock.hcl", "terraform.tfstate"])
def _root_dart(file_path: str, workspace: str) -> Optional[str]:
return _root_or_workspace(file_path, workspace, ["pubspec.yaml", "analysis_options.yaml"])
def _root_haskell(file_path: str, workspace: str) -> Optional[str]:
return _root_or_workspace(file_path, workspace, ["stack.yaml", "cabal.project", "hie.yaml"])
def _root_julia(file_path: str, workspace: str) -> Optional[str]:
return _root_or_workspace(file_path, workspace, ["Project.toml", "Manifest.toml"])
def _root_clojure(file_path: str, workspace: str) -> Optional[str]:
return _root_or_workspace(
file_path, workspace, ["deps.edn", "project.clj", "shadow-cljs.edn", "bb.edn", "build.boot"]
)
def _root_nix(file_path: str, workspace: str) -> str:
found = nearest_root(file_path, ["flake.nix"])
return found or workspace
def _root_zig(file_path: str, workspace: str) -> Optional[str]:
return _root_or_workspace(file_path, workspace, ["build.zig"])
def _root_elixir(file_path: str, workspace: str) -> Optional[str]:
return _root_or_workspace(file_path, workspace, ["mix.exs", "mix.lock"])
def _root_prisma(file_path: str, workspace: str) -> Optional[str]:
return _root_or_workspace(
file_path, workspace, ["schema.prisma", "prisma/schema.prisma"]
)
def _root_kotlin(file_path: str, workspace: str) -> Optional[str]:
return _root_or_workspace(
file_path,
workspace,
["settings.gradle", "settings.gradle.kts", "build.gradle", "build.gradle.kts", "pom.xml"],
)
def _root_java(file_path: str, workspace: str) -> Optional[str]:
return _root_or_workspace(
file_path,
workspace,
["pom.xml", "build.gradle", "build.gradle.kts", ".project", ".classpath", "settings.gradle"],
)
# ---------------------------------------------------------------------------
# the registry
# ---------------------------------------------------------------------------
SERVERS: List[ServerDef] = [
ServerDef(
server_id="pyright",
extensions=(".py", ".pyi"),
resolve_root=_root_python,
build_spawn=_spawn_pyright,
description="Python — Microsoft pyright",
),
ServerDef(
server_id="typescript",
extensions=(".ts", ".tsx", ".js", ".jsx", ".mjs", ".cjs", ".mts", ".cts"),
resolve_root=_root_typescript,
build_spawn=_spawn_typescript,
seed_first_push=True,
description="JavaScript/TypeScript — typescript-language-server",
),
ServerDef(
server_id="vue-language-server",
extensions=(".vue",),
resolve_root=_root_typescript,
build_spawn=_spawn_vue,
description="Vue.js — @vue/language-server",
),
ServerDef(
server_id="svelte-language-server",
extensions=(".svelte",),
resolve_root=_root_typescript,
build_spawn=_spawn_svelte,
description="Svelte — svelte-language-server",
),
ServerDef(
server_id="astro-language-server",
extensions=(".astro",),
resolve_root=_root_typescript,
build_spawn=_spawn_astro,
description="Astro — @astrojs/language-server",
),
ServerDef(
server_id="gopls",
extensions=(".go",),
resolve_root=_root_go,
build_spawn=_spawn_gopls,
description="Go — gopls",
),
ServerDef(
server_id="rust-analyzer",
extensions=(".rs",),
resolve_root=_root_rust,
build_spawn=_spawn_rust_analyzer,
description="Rust — rust-analyzer",
),
ServerDef(
server_id="clangd",
extensions=(".c", ".cpp", ".cc", ".cxx", ".h", ".hh", ".hpp", ".hxx"),
resolve_root=_root_clangd,
build_spawn=_spawn_clangd,
description="C/C++ — clangd",
),
ServerDef(
server_id="bash-language-server",
extensions=(".sh", ".bash", ".zsh", ".ksh"),
resolve_root=_root_bash,
build_spawn=_spawn_bash_ls,
description="Bash — bash-language-server",
),
ServerDef(
server_id="yaml-language-server",
extensions=(".yaml", ".yml"),
resolve_root=_root_yaml,
build_spawn=_spawn_yaml_ls,
description="YAML — yaml-language-server",
),
ServerDef(
server_id="lua-language-server",
extensions=(".lua",),
resolve_root=_root_lua,
build_spawn=_spawn_lua_ls,
description="Lua — lua-language-server",
),
ServerDef(
server_id="intelephense",
extensions=(".php",),
resolve_root=_root_php,
build_spawn=_spawn_intelephense,
description="PHP — intelephense",
),
ServerDef(
server_id="ocaml-lsp",
extensions=(".ml", ".mli"),
resolve_root=_root_ocaml,
build_spawn=_spawn_ocamllsp,
description="OCaml — ocaml-lsp",
),
ServerDef(
server_id="dockerfile-ls",
extensions=(".dockerfile", "Dockerfile"),
resolve_root=_root_docker,
build_spawn=_spawn_dockerfile_ls,
description="Dockerfile — dockerfile-language-server-nodejs",
),
ServerDef(
server_id="terraform-ls",
extensions=(".tf", ".tfvars"),
resolve_root=_root_terraform,
build_spawn=_spawn_terraform_ls,
description="Terraform — terraform-ls",
),
ServerDef(
server_id="dart",
extensions=(".dart",),
resolve_root=_root_dart,
build_spawn=_spawn_dart,
description="Dart — built-in language server",
),
ServerDef(
server_id="haskell-language-server",
extensions=(".hs", ".lhs"),
resolve_root=_root_haskell,
build_spawn=_spawn_haskell_ls,
description="Haskell — haskell-language-server",
),
ServerDef(
server_id="julia",
extensions=(".jl",),
resolve_root=_root_julia,
build_spawn=_spawn_julia,
description="Julia — LanguageServer.jl",
),
ServerDef(
server_id="clojure-lsp",
extensions=(".clj", ".cljs", ".cljc", ".edn"),
resolve_root=_root_clojure,
build_spawn=_spawn_clojure_lsp,
description="Clojure — clojure-lsp",
),
ServerDef(
server_id="nixd",
extensions=(".nix",),
resolve_root=_root_nix,
build_spawn=_spawn_nixd,
description="Nix — nixd",
),
ServerDef(
server_id="zls",
extensions=(".zig", ".zon"),
resolve_root=_root_zig,
build_spawn=_spawn_zls,
description="Zig — zls",
),
ServerDef(
server_id="gleam",
extensions=(".gleam",),
resolve_root=lambda fp, ws: _root_or_workspace(fp, ws, ["gleam.toml"]),
build_spawn=_spawn_gleam,
description="Gleam — built-in language server",
),
ServerDef(
server_id="elixir-ls",
extensions=(".ex", ".exs"),
resolve_root=_root_elixir,
build_spawn=_spawn_elixir_ls,
description="Elixir — elixir-ls",
),
ServerDef(
server_id="prisma",
extensions=(".prisma",),
resolve_root=_root_prisma,
build_spawn=_spawn_prisma,
description="Prisma — built-in language server",
),
ServerDef(
server_id="kotlin-language-server",
extensions=(".kt", ".kts"),
resolve_root=_root_kotlin,
build_spawn=_spawn_kotlin_ls,
description="Kotlin — kotlin-language-server",
),
ServerDef(
server_id="jdtls",
extensions=(".java",),
resolve_root=_root_java,
build_spawn=_spawn_jdtls,
description="Java — Eclipse JDT Language Server",
),
]
def find_server_for_file(file_path: str) -> Optional[ServerDef]:
"""Return the registry entry that handles ``file_path``, or None."""
for srv in SERVERS:
if srv.matches(file_path):
return srv
return None
def language_id_for(path: str) -> str:
"""Return the LSP languageId to send in didOpen for ``path``."""
ext = _file_ext_or_basename(path)
return LANGUAGE_BY_EXT.get(ext, "plaintext")
__all__ = [
"ServerDef",
"ServerContext",
"SpawnSpec",
"SERVERS",
"find_server_for_file",
"language_id_for",
"LANGUAGE_BY_EXT",
]