mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-22 16:25:58 +00:00
eb09df6ec8
The staleness fix (f9b1fd799) bolted two wall-clock dicts (_changed_at, _pulled_at) onto a client that already scattered per-document state across six parallel dicts (_files, _push_diagnostics, _pull_diagnostics, _published, _published_version, _first_push_seen) — eight maps kept in sync by hand. Collapse all of it into one _DocState per path, and use the LSP document version as the freshness token instead of clocks: - didChange bumps doc.version; stored push/pull results carry the version they describe (push_version from the server's echoed version, or the current version at receipt for servers that don't echo one; pull_version captured at request send so an in-flight pull that a didChange races past is stale on arrival). - fresh == tag >= version. Invalidation is implicit in the bump — no store-clearing, no clock comparisons, no race windows. - _has_fresh_push/_has_fresh_pull helpers dissolve into two one-line _DocState methods; diagnostics_for(fresh_only=True) becomes a three-liner. Semantics are unchanged fromf9b1fd799(same tests pass, one test updated off private internals); net -15 lines.
1020 lines
41 KiB
Python
1020 lines
41 KiB
Python
"""Async LSP client over stdin/stdout.
|
|
|
|
One :class:`LSPClient` corresponds to one ``(language_server, workspace_root)``
|
|
pair — exactly what OpenCode keys clients on, and the same shape Claude
|
|
Code uses. The client owns a child process, drives the JSON-RPC
|
|
exchange, and exposes:
|
|
|
|
- :meth:`open_file` / :meth:`change_file` — text document sync
|
|
- :meth:`wait_for_diagnostics` — block until the server emits fresh
|
|
diagnostics for a specific file (or a timeout fires)
|
|
- :meth:`diagnostics_for` — read the current per-file diagnostic store
|
|
- :meth:`shutdown` — graceful close + SIGTERM/SIGKILL fallback
|
|
|
|
The class is designed for async use from a single asyncio event loop.
|
|
The :class:`agent.lsp.manager.LSPService` runs an event loop in a
|
|
background thread so the synchronous file_operations layer can call
|
|
into it via :func:`agent.lsp.manager.LSPService.touch_file`.
|
|
|
|
Implementation notes:
|
|
|
|
- All per-document state lives in one :class:`_DocState` keyed by
|
|
absolute path. Freshness is tracked with **document versions**,
|
|
not timestamps: every didChange bumps ``version``, and each stored
|
|
push/pull result is tagged with the version it describes. A
|
|
result is fresh iff its tag >= the version being waited on, so a
|
|
didChange implicitly invalidates everything older — no clearing,
|
|
no clock comparisons, no race windows. This is what prevents
|
|
"ghost diagnostics": a slow server's leftovers from the previous
|
|
edit can never masquerade as a verdict on the current content.
|
|
|
|
- Whole-document sync. Even when the server advertises incremental
|
|
sync, we send a single ``contentChanges`` entry replacing the
|
|
entire document. Pretending to be incremental while sending a
|
|
full replacement is well-tolerated by every major server and saves
|
|
range bookkeeping. See OpenCode's ``client.ts:584-659`` for the
|
|
same trick.
|
|
|
|
- The "touch-file dance": every ``open_file`` call also fires a
|
|
``workspace/didChangeWatchedFiles`` notification (CREATED on the
|
|
first open, CHANGED thereafter). Some servers (clangd, eslint)
|
|
only re-scan when this notification fires, even though the LSP spec
|
|
doesn't strictly require it.
|
|
|
|
- ``ContentModified`` (-32801) errors get retried with exponential
|
|
backoff up to 3 times. This matches Claude Code's
|
|
``LSPServerInstance.sendRequest``.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import sys
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any, Awaitable, Callable, Dict, List, Optional, Set
|
|
from urllib.parse import quote, unquote
|
|
|
|
from agent.lsp.protocol import (
|
|
ERROR_CONTENT_MODIFIED,
|
|
ERROR_METHOD_NOT_FOUND,
|
|
LSPProtocolError,
|
|
LSPRequestError,
|
|
classify_message,
|
|
encode_message,
|
|
make_error_response,
|
|
make_notification,
|
|
make_request,
|
|
make_response,
|
|
read_message,
|
|
)
|
|
|
|
logger = logging.getLogger("agent.lsp.client")
|
|
|
|
# Timeouts (seconds) — mirror OpenCode's constants, scaled to seconds.
|
|
INITIALIZE_TIMEOUT = 45.0
|
|
DIAGNOSTICS_DOCUMENT_WAIT = 5.0
|
|
DIAGNOSTICS_FULL_WAIT = 10.0
|
|
DIAGNOSTICS_REQUEST_TIMEOUT = 3.0
|
|
PUSH_DEBOUNCE = 0.15
|
|
SHUTDOWN_GRACE = 1.0 # seconds between SIGTERM and SIGKILL
|
|
|
|
# Retry policy for transient ContentModified errors.
|
|
MAX_CONTENT_MODIFIED_RETRIES = 3
|
|
RETRY_BASE_DELAY = 0.5 # 0.5, 1.0, 2.0 — exponential
|
|
|
|
|
|
def file_uri(path: str) -> str:
|
|
"""Return ``file://`` URI for an absolute filesystem path.
|
|
|
|
Mirrors Node's ``pathToFileURL`` — handles spaces, unicode, and
|
|
Windows drive letters (``C:\\foo`` → ``file:///C:/foo``).
|
|
"""
|
|
abs_path = os.path.abspath(path)
|
|
if os.name == "nt":
|
|
# Windows: backslash → forward slash, prepend extra slash so
|
|
# the drive letter shows up as part of the path component.
|
|
abs_path = abs_path.replace("\\", "/")
|
|
if not abs_path.startswith("/"):
|
|
abs_path = "/" + abs_path
|
|
return "file://" + quote(abs_path, safe="/:")
|
|
|
|
|
|
def uri_to_path(uri: str) -> str:
|
|
"""Inverse of :func:`file_uri`."""
|
|
if not uri.startswith("file://"):
|
|
return uri
|
|
raw = uri[len("file://"):]
|
|
if os.name == "nt" and raw.startswith("/") and len(raw) > 2 and raw[2] == ":":
|
|
raw = raw[1:] # strip leading slash before drive letter
|
|
return os.path.normpath(unquote(raw))
|
|
|
|
|
|
def _end_position(text: str) -> Dict[str, int]:
|
|
"""Return the LSP Position at the end of ``text``.
|
|
|
|
Used to construct a single-range "replace whole document" change
|
|
for ``textDocument/didChange`` regardless of the server's declared
|
|
sync mode.
|
|
"""
|
|
if not text:
|
|
return {"line": 0, "character": 0}
|
|
lines = text.splitlines(keepends=False)
|
|
last_line = len(lines) - 1
|
|
last_col = len(lines[-1]) if lines else 0
|
|
# If the text ends with a trailing newline, ``splitlines`` won't
|
|
# represent it. The end position is then the start of the next
|
|
# (empty) line — line index is len(lines), column 0.
|
|
if text.endswith(("\n", "\r")):
|
|
return {"line": last_line + 1, "character": 0}
|
|
return {"line": last_line, "character": last_col}
|
|
|
|
|
|
@dataclass
|
|
class _DocState:
|
|
"""Everything the client tracks for one open document.
|
|
|
|
``version`` is the LSP document version we last sent (didOpen=0,
|
|
each didChange +1). It doubles as the freshness token: stored
|
|
push/pull results are tagged with the version they describe
|
|
(``push_version`` / ``pull_version``), and a result is *fresh*
|
|
iff its tag has caught up to ``version``. Bumping the version on
|
|
didChange therefore invalidates all older results implicitly —
|
|
no store-clearing, no timestamps.
|
|
|
|
``push_version``/``pull_version`` start at -1 = "no data yet".
|
|
Servers that echo a document version in publishDiagnostics get
|
|
exact tagging; those that don't are credited with the current
|
|
version at receipt time (a push observed after we sent the
|
|
change describes the changed content or newer).
|
|
"""
|
|
|
|
version: int = 0
|
|
text: str = ""
|
|
push: List[Dict[str, Any]] = field(default_factory=list)
|
|
pull: List[Dict[str, Any]] = field(default_factory=list)
|
|
push_version: int = -1
|
|
pull_version: int = -1
|
|
seed_seen: bool = False
|
|
|
|
def fresh_push(self, version: Optional[int] = None) -> bool:
|
|
return self.push_version >= (self.version if version is None else version)
|
|
|
|
def fresh_pull(self, version: Optional[int] = None) -> bool:
|
|
return self.pull_version >= (self.version if version is None else version)
|
|
|
|
|
|
class LSPClient:
|
|
"""Async LSP client tied to one server process and one workspace root.
|
|
|
|
Lifecycle:
|
|
|
|
c = LSPClient(server_id, workspace_root, command, args, init_options)
|
|
await c.start() # spawn + initialize
|
|
ver = await c.open_file("/path/to/foo.py")
|
|
await c.wait_for_diagnostics("/path/to/foo.py", ver)
|
|
diags = c.diagnostics_for("/path/to/foo.py")
|
|
await c.shutdown()
|
|
"""
|
|
|
|
# ------------------------------------------------------------------
|
|
# construction + lifecycle
|
|
# ------------------------------------------------------------------
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
server_id: str,
|
|
workspace_root: str,
|
|
command: List[str],
|
|
env: Optional[Dict[str, str]] = None,
|
|
cwd: Optional[str] = None,
|
|
initialization_options: Optional[Dict[str, Any]] = None,
|
|
seed_diagnostics_on_first_push: bool = False,
|
|
) -> None:
|
|
self.server_id = server_id
|
|
self.workspace_root = workspace_root
|
|
self._command = list(command)
|
|
self._env = env
|
|
self._cwd = cwd or workspace_root
|
|
self._init_options = initialization_options or {}
|
|
self._seed_first_push = seed_diagnostics_on_first_push
|
|
|
|
# Process + streams
|
|
self._proc: Optional[asyncio.subprocess.Process] = None
|
|
self._stderr_task: Optional[asyncio.Task] = None
|
|
self._reader_task: Optional[asyncio.Task] = None
|
|
|
|
# Request/response correlation
|
|
self._next_id: int = 0
|
|
self._pending: Dict[int, asyncio.Future] = {}
|
|
|
|
# Server-side request handlers (server → client requests).
|
|
# Kept small and explicit; everything else returns method-not-found.
|
|
self._request_handlers: Dict[str, Callable[[Any], Awaitable[Any]]] = {
|
|
"window/workDoneProgress/create": self._handle_work_done_create,
|
|
"workspace/configuration": self._handle_workspace_configuration,
|
|
"client/registerCapability": self._handle_register_capability,
|
|
"client/unregisterCapability": self._handle_unregister_capability,
|
|
"workspace/workspaceFolders": self._handle_workspace_folders,
|
|
"workspace/diagnostic/refresh": self._handle_diagnostic_refresh,
|
|
}
|
|
# Notifications (server → client) we care about.
|
|
self._notification_handlers: Dict[str, Callable[[Any], None]] = {
|
|
"textDocument/publishDiagnostics": self._handle_publish_diagnostics,
|
|
# Everything else (window/showMessage, $/progress, etc.)
|
|
# is silently dropped by default.
|
|
}
|
|
|
|
# Per-document state (version, text, diagnostic stores, and
|
|
# their freshness tags), keyed by absolute file path (NOT URI).
|
|
# See _DocState for the version-based freshness model.
|
|
self._docs: Dict[str, _DocState] = {}
|
|
# Capability registrations — only diagnostic ones are tracked.
|
|
self._diagnostic_registrations: Dict[str, Dict[str, Any]] = {}
|
|
|
|
# State machine
|
|
self._state: str = "stopped"
|
|
self._initialize_result: Optional[Dict[str, Any]] = None
|
|
self._sync_kind: int = 1 # 1=Full, 2=Incremental
|
|
self._stopping: bool = False
|
|
|
|
# Push event for waiters.
|
|
self._push_event = asyncio.Event()
|
|
# Monotonic counter incremented on every publishDiagnostics push.
|
|
# Waiters snapshot it on entry and treat any increase as
|
|
# "something happened, recheck the predicate". Avoids the
|
|
# asyncio.Event sticky-state trap.
|
|
self._push_counter = 0
|
|
# Registration change event so wait_for_diagnostics can re-loop
|
|
# when the server announces a new dynamic provider.
|
|
self._registration_event = asyncio.Event()
|
|
|
|
@property
|
|
def is_running(self) -> bool:
|
|
return self._state == "running" and self._proc is not None and self._proc.returncode is None
|
|
|
|
@property
|
|
def state(self) -> str:
|
|
return self._state
|
|
|
|
async def start(self) -> None:
|
|
"""Spawn the server and complete the initialize handshake.
|
|
|
|
Raises any exception encountered during spawn/init. On failure
|
|
the process is killed and the client is left in state
|
|
``"error"`` — re-call ``start()`` to retry.
|
|
"""
|
|
if self._state in {"running", "starting"}:
|
|
return
|
|
self._state = "starting"
|
|
try:
|
|
await self._spawn()
|
|
await self._initialize()
|
|
self._state = "running"
|
|
except Exception:
|
|
self._state = "error"
|
|
await self._cleanup_process()
|
|
raise
|
|
|
|
@staticmethod
|
|
def _win_wrap_cmd(cmd: List[str]) -> List[str]:
|
|
"""On Windows, wrap .cmd/.bat shims so CreateProcess can run them."""
|
|
exe = cmd[0]
|
|
if exe.lower().endswith((".cmd", ".bat")):
|
|
return ["cmd.exe", "/c", *cmd]
|
|
return cmd
|
|
|
|
async def _spawn(self) -> None:
|
|
env = dict(os.environ)
|
|
if self._env:
|
|
env.update(self._env)
|
|
|
|
cmd = self._command
|
|
if sys.platform == "win32":
|
|
cmd = self._win_wrap_cmd(cmd)
|
|
|
|
try:
|
|
# start_new_session=True detaches the LSP server into its own
|
|
# process group / session. Without this, the LSP server inherits
|
|
# the gateway's pgid (= TUI parent PID). When mcp_tool's
|
|
# _kill_orphaned_mcp_children races with LSP spawn and sweeps the
|
|
# gateway's child set, it captures the LSP PID, records the
|
|
# inherited pgid, and killpg() then kills the TUI parent itself.
|
|
# See tui_gateway_crash.log "killpg → SIGTERM received" stacks.
|
|
self._proc = await asyncio.create_subprocess_exec(
|
|
cmd[0],
|
|
*cmd[1:],
|
|
stdin=asyncio.subprocess.PIPE,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
env=env,
|
|
cwd=self._cwd,
|
|
start_new_session=True,
|
|
)
|
|
except FileNotFoundError as e:
|
|
raise LSPProtocolError(
|
|
f"LSP server binary not found: {cmd[0]} ({e})"
|
|
) from e
|
|
|
|
# Drain stderr at debug level — if we don't, the pipe buffer
|
|
# fills and the server hangs.
|
|
self._stderr_task = asyncio.create_task(self._drain_stderr())
|
|
# Start the reader loop.
|
|
self._reader_task = asyncio.create_task(self._reader_loop())
|
|
|
|
async def _drain_stderr(self) -> None:
|
|
if self._proc is None or self._proc.stderr is None:
|
|
return
|
|
try:
|
|
while True:
|
|
line = await self._proc.stderr.readline()
|
|
if not line:
|
|
break
|
|
text = line.decode("utf-8", errors="replace").rstrip()
|
|
if text:
|
|
logger.debug("[%s] stderr: %s", self.server_id, text[:1000])
|
|
except (asyncio.CancelledError, OSError):
|
|
pass
|
|
|
|
async def _reader_loop(self) -> None:
|
|
if self._proc is None or self._proc.stdout is None:
|
|
return
|
|
try:
|
|
while True:
|
|
msg = await read_message(self._proc.stdout)
|
|
if msg is None:
|
|
logger.debug("[%s] server closed stdout cleanly", self.server_id)
|
|
break
|
|
kind, key = classify_message(msg)
|
|
if kind == "response":
|
|
self._dispatch_response(key, msg)
|
|
elif kind == "request":
|
|
asyncio.create_task(self._dispatch_request(key, msg))
|
|
elif kind == "notification":
|
|
self._dispatch_notification(key, msg)
|
|
else:
|
|
logger.warning("[%s] dropping invalid message: %r", self.server_id, msg)
|
|
except LSPProtocolError as e:
|
|
logger.warning("[%s] protocol error in reader loop: %s", self.server_id, e)
|
|
except (asyncio.CancelledError, OSError):
|
|
pass
|
|
finally:
|
|
# Wake up any pending requests so they can fail fast.
|
|
for fut in list(self._pending.values()):
|
|
if not fut.done():
|
|
fut.set_exception(LSPProtocolError("server connection closed"))
|
|
self._pending.clear()
|
|
|
|
async def _initialize(self) -> None:
|
|
params = {
|
|
"rootUri": file_uri(self.workspace_root),
|
|
"rootPath": self.workspace_root,
|
|
"processId": os.getpid(),
|
|
"workspaceFolders": [
|
|
{"name": "workspace", "uri": file_uri(self.workspace_root)}
|
|
],
|
|
"initializationOptions": self._init_options,
|
|
"capabilities": {
|
|
"window": {"workDoneProgress": True},
|
|
"workspace": {
|
|
"configuration": True,
|
|
"workspaceFolders": True,
|
|
"didChangeWatchedFiles": {"dynamicRegistration": True},
|
|
"diagnostics": {"refreshSupport": False},
|
|
},
|
|
"textDocument": {
|
|
"synchronization": {
|
|
"dynamicRegistration": False,
|
|
"didOpen": True,
|
|
"didChange": True,
|
|
"didSave": True,
|
|
"willSave": False,
|
|
"willSaveWaitUntil": False,
|
|
},
|
|
"diagnostic": {
|
|
"dynamicRegistration": True,
|
|
"relatedDocumentSupport": True,
|
|
},
|
|
"publishDiagnostics": {
|
|
"relatedInformation": True,
|
|
"tagSupport": {"valueSet": [1, 2]},
|
|
"versionSupport": True,
|
|
"codeDescriptionSupport": True,
|
|
"dataSupport": False,
|
|
},
|
|
"hover": {"contentFormat": ["markdown", "plaintext"]},
|
|
"definition": {"linkSupport": True},
|
|
"references": {},
|
|
"documentSymbol": {"hierarchicalDocumentSymbolSupport": True},
|
|
},
|
|
"general": {"positionEncodings": ["utf-16"]},
|
|
},
|
|
}
|
|
|
|
result = await asyncio.wait_for(
|
|
self._send_request("initialize", params),
|
|
timeout=INITIALIZE_TIMEOUT,
|
|
)
|
|
self._initialize_result = result
|
|
self._sync_kind = self._extract_sync_kind(result.get("capabilities") or {})
|
|
|
|
await self._send_notification("initialized", {})
|
|
if self._init_options:
|
|
# Some servers (vtsls, eslint) want config pushed via
|
|
# didChangeConfiguration even if it was sent in
|
|
# initializationOptions.
|
|
await self._send_notification(
|
|
"workspace/didChangeConfiguration",
|
|
{"settings": self._init_options},
|
|
)
|
|
|
|
@staticmethod
|
|
def _extract_sync_kind(capabilities: dict) -> int:
|
|
sync = capabilities.get("textDocumentSync")
|
|
if isinstance(sync, int):
|
|
return sync
|
|
if isinstance(sync, dict):
|
|
change = sync.get("change")
|
|
if isinstance(change, int):
|
|
return change
|
|
return 1 # default to Full
|
|
|
|
async def shutdown(self) -> None:
|
|
"""Best-effort graceful shutdown.
|
|
|
|
Sends ``shutdown`` + ``exit``, then SIGTERMs/SIGKILLs the
|
|
process if it doesn't exit cleanly. Idempotent.
|
|
"""
|
|
if self._stopping:
|
|
return
|
|
self._stopping = True
|
|
try:
|
|
if self.is_running:
|
|
try:
|
|
await asyncio.wait_for(self._send_request("shutdown", None), timeout=2.0)
|
|
except (asyncio.TimeoutError, LSPRequestError, LSPProtocolError):
|
|
pass
|
|
try:
|
|
await self._send_notification("exit", None)
|
|
except Exception:
|
|
pass
|
|
finally:
|
|
self._state = "stopped"
|
|
await self._cleanup_process()
|
|
|
|
async def _cleanup_process(self) -> None:
|
|
if self._reader_task is not None and not self._reader_task.done():
|
|
self._reader_task.cancel()
|
|
try:
|
|
await self._reader_task
|
|
except (asyncio.CancelledError, Exception): # noqa: BLE001
|
|
pass
|
|
if self._stderr_task is not None and not self._stderr_task.done():
|
|
self._stderr_task.cancel()
|
|
try:
|
|
await self._stderr_task
|
|
except (asyncio.CancelledError, Exception): # noqa: BLE001
|
|
pass
|
|
proc = self._proc
|
|
self._proc = None
|
|
if proc is None:
|
|
return
|
|
if proc.returncode is None:
|
|
try:
|
|
proc.terminate()
|
|
try:
|
|
await asyncio.wait_for(proc.wait(), timeout=SHUTDOWN_GRACE)
|
|
except asyncio.TimeoutError:
|
|
try:
|
|
proc.kill()
|
|
await proc.wait()
|
|
except ProcessLookupError:
|
|
pass
|
|
except ProcessLookupError:
|
|
pass
|
|
|
|
# ------------------------------------------------------------------
|
|
# request / notification plumbing
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _send_request(self, method: str, params: Any) -> Any:
|
|
if self._proc is None or self._proc.stdin is None or self._proc.stdin.is_closing():
|
|
raise LSPProtocolError(f"cannot send {method!r}: stdin closed")
|
|
loop = asyncio.get_running_loop()
|
|
req_id = self._next_id
|
|
self._next_id += 1
|
|
fut: asyncio.Future = loop.create_future()
|
|
self._pending[req_id] = fut
|
|
try:
|
|
self._proc.stdin.write(encode_message(make_request(req_id, method, params)))
|
|
await self._proc.stdin.drain()
|
|
except (BrokenPipeError, ConnectionResetError, OSError) as e:
|
|
self._pending.pop(req_id, None)
|
|
raise LSPProtocolError(f"send failed for {method!r}: {e}") from e
|
|
try:
|
|
return await fut
|
|
finally:
|
|
self._pending.pop(req_id, None)
|
|
|
|
async def _send_request_with_retry(self, method: str, params: Any, *, timeout: float) -> Any:
|
|
"""Send a request, retrying on ``ContentModified`` (-32801).
|
|
|
|
Other errors propagate. The retry policy matches Claude Code's
|
|
``LSPServerInstance.sendRequest`` — 3 attempts with delays
|
|
0.5s, 1.0s, 2.0s.
|
|
"""
|
|
for attempt in range(MAX_CONTENT_MODIFIED_RETRIES + 1):
|
|
try:
|
|
return await asyncio.wait_for(self._send_request(method, params), timeout=timeout)
|
|
except LSPRequestError as e:
|
|
if e.code == ERROR_CONTENT_MODIFIED and attempt < MAX_CONTENT_MODIFIED_RETRIES:
|
|
await asyncio.sleep(RETRY_BASE_DELAY * (2 ** attempt))
|
|
continue
|
|
raise
|
|
|
|
async def _send_notification(self, method: str, params: Any) -> None:
|
|
if self._proc is None or self._proc.stdin is None or self._proc.stdin.is_closing():
|
|
return
|
|
try:
|
|
self._proc.stdin.write(encode_message(make_notification(method, params)))
|
|
await self._proc.stdin.drain()
|
|
except (BrokenPipeError, ConnectionResetError, OSError) as e:
|
|
logger.debug("[%s] notify %s failed: %s", self.server_id, method, e)
|
|
|
|
async def _send_response(self, req_id: Any, result: Any) -> None:
|
|
if self._proc is None or self._proc.stdin is None or self._proc.stdin.is_closing():
|
|
return
|
|
try:
|
|
self._proc.stdin.write(encode_message(make_response(req_id, result)))
|
|
await self._proc.stdin.drain()
|
|
except (BrokenPipeError, ConnectionResetError, OSError):
|
|
pass
|
|
|
|
async def _send_error_response(self, req_id: Any, code: int, message: str) -> None:
|
|
if self._proc is None or self._proc.stdin is None or self._proc.stdin.is_closing():
|
|
return
|
|
try:
|
|
self._proc.stdin.write(encode_message(make_error_response(req_id, code, message)))
|
|
await self._proc.stdin.drain()
|
|
except (BrokenPipeError, ConnectionResetError, OSError):
|
|
pass
|
|
|
|
def _dispatch_response(self, req_id: int, msg: dict) -> None:
|
|
fut = self._pending.get(req_id)
|
|
if fut is None or fut.done():
|
|
return
|
|
if "error" in msg:
|
|
err = msg["error"] or {}
|
|
fut.set_exception(
|
|
LSPRequestError(
|
|
code=int(err.get("code", -32000)),
|
|
message=str(err.get("message", "unknown")),
|
|
data=err.get("data"),
|
|
)
|
|
)
|
|
else:
|
|
fut.set_result(msg.get("result"))
|
|
|
|
async def _dispatch_request(self, req_id: Any, msg: dict) -> None:
|
|
method = msg.get("method", "")
|
|
params = msg.get("params")
|
|
handler = self._request_handlers.get(method)
|
|
if handler is None:
|
|
await self._send_error_response(req_id, ERROR_METHOD_NOT_FOUND, f"method not found: {method}")
|
|
return
|
|
try:
|
|
result = await handler(params)
|
|
except Exception as e: # noqa: BLE001 — protocol must not blow up
|
|
logger.warning("[%s] request handler %s failed: %s", self.server_id, method, e)
|
|
await self._send_error_response(req_id, -32000, f"handler failed: {e}")
|
|
return
|
|
await self._send_response(req_id, result)
|
|
|
|
def _dispatch_notification(self, method: str, msg: dict) -> None:
|
|
handler = self._notification_handlers.get(method)
|
|
if handler is None:
|
|
return
|
|
try:
|
|
handler(msg.get("params"))
|
|
except Exception as e: # noqa: BLE001
|
|
logger.debug("[%s] notification handler %s failed: %s", self.server_id, method, e)
|
|
|
|
# ------------------------------------------------------------------
|
|
# built-in server-→-client request handlers
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _handle_work_done_create(self, params: Any) -> Any:
|
|
# Acknowledge progress tokens — required by some servers.
|
|
return None
|
|
|
|
async def _handle_workspace_configuration(self, params: Any) -> Any:
|
|
# Walk dotted sections through initializationOptions. Mirrors
|
|
# OpenCode's `client.ts:198-220` — return null when missing.
|
|
if not isinstance(params, dict):
|
|
return [None]
|
|
items = params.get("items") or []
|
|
out: List[Any] = []
|
|
for item in items:
|
|
if not isinstance(item, dict):
|
|
out.append(None)
|
|
continue
|
|
section = item.get("section")
|
|
if not section or not self._init_options:
|
|
out.append(self._init_options or None)
|
|
continue
|
|
cur: Any = self._init_options
|
|
for part in str(section).split("."):
|
|
if isinstance(cur, dict) and part in cur:
|
|
cur = cur[part]
|
|
else:
|
|
cur = None
|
|
break
|
|
out.append(cur)
|
|
return out
|
|
|
|
async def _handle_register_capability(self, params: Any) -> Any:
|
|
if not isinstance(params, dict):
|
|
return None
|
|
for reg in params.get("registrations") or []:
|
|
if not isinstance(reg, dict):
|
|
continue
|
|
method = reg.get("method")
|
|
reg_id = reg.get("id")
|
|
if method == "textDocument/diagnostic" and reg_id:
|
|
self._diagnostic_registrations[str(reg_id)] = reg
|
|
self._registration_event.set()
|
|
return None
|
|
|
|
async def _handle_unregister_capability(self, params: Any) -> Any:
|
|
if not isinstance(params, dict):
|
|
return None
|
|
for unreg in params.get("unregisterations") or []:
|
|
if not isinstance(unreg, dict):
|
|
continue
|
|
reg_id = unreg.get("id")
|
|
if reg_id:
|
|
self._diagnostic_registrations.pop(str(reg_id), None)
|
|
return None
|
|
|
|
async def _handle_workspace_folders(self, params: Any) -> Any:
|
|
return [{"name": "workspace", "uri": file_uri(self.workspace_root)}]
|
|
|
|
async def _handle_diagnostic_refresh(self, params: Any) -> Any:
|
|
# We don't honour refresh — we re-pull on every touchFile.
|
|
return None
|
|
|
|
# ------------------------------------------------------------------
|
|
# publishDiagnostics handler
|
|
# ------------------------------------------------------------------
|
|
|
|
def _handle_publish_diagnostics(self, params: Any) -> None:
|
|
if not isinstance(params, dict):
|
|
return
|
|
uri = params.get("uri")
|
|
if not isinstance(uri, str):
|
|
return
|
|
path = uri_to_path(uri)
|
|
diagnostics = params.get("diagnostics") or []
|
|
if not isinstance(diagnostics, list):
|
|
diagnostics = []
|
|
version = params.get("version")
|
|
|
|
doc = self._docs.setdefault(path, _DocState(version=-1))
|
|
if self._seed_first_push and not doc.seed_seen:
|
|
# First push: seed the store WITHOUT a freshness tag. It
|
|
# arrives before the user-triggered didChange could've
|
|
# produced fresh diagnostics, so it must never satisfy a
|
|
# waiter — it's baseline data only.
|
|
doc.seed_seen = True
|
|
doc.push = diagnostics
|
|
return
|
|
|
|
doc.seed_seen = True
|
|
doc.push = diagnostics
|
|
# Tag with the echoed document version when the server provides
|
|
# one; otherwise credit the current version — a push observed
|
|
# after we sent the change describes the changed content (or
|
|
# newer). Note doc.version is -1 for never-opened paths
|
|
# (e.g. relatedDocuments spillover), keeping them unfresh.
|
|
doc.push_version = version if isinstance(version, int) else doc.version
|
|
# Bump the monotonic push counter and wake every waiter. We
|
|
# keep the Event sticky-set so any wait already in progress
|
|
# resolves; waiters re-check their predicate after waking and
|
|
# decide whether to keep waiting. ``_push_counter`` is what
|
|
# they actually compare against to detect a fresh event.
|
|
self._push_counter += 1
|
|
self._push_event.set()
|
|
|
|
# ------------------------------------------------------------------
|
|
# public file-sync API
|
|
# ------------------------------------------------------------------
|
|
|
|
async def open_file(self, path: str, *, language_id: str = "plaintext") -> int:
|
|
"""Send didOpen (first time) or didChange (subsequent) for ``path``.
|
|
|
|
Returns the new document version number that the agent's
|
|
``wait_for_diagnostics`` should match against.
|
|
"""
|
|
if not self.is_running:
|
|
raise LSPProtocolError("client not running")
|
|
|
|
abs_path = os.path.abspath(path)
|
|
try:
|
|
text = Path(abs_path).read_text(encoding="utf-8", errors="replace")
|
|
except OSError as e:
|
|
raise LSPProtocolError(f"cannot read {abs_path}: {e}") from e
|
|
|
|
uri = file_uri(abs_path)
|
|
doc = self._docs.get(abs_path)
|
|
|
|
if doc is not None and doc.version >= 0:
|
|
# Re-open: bump version, fire didChangeWatchedFiles + didChange.
|
|
await self._send_notification(
|
|
"workspace/didChangeWatchedFiles",
|
|
{"changes": [{"uri": uri, "type": 2}]}, # 2 = CHANGED
|
|
)
|
|
new_version = doc.version + 1
|
|
old_text = doc.text
|
|
content_changes: List[Dict[str, Any]]
|
|
if self._sync_kind == 2:
|
|
content_changes = [
|
|
{
|
|
"range": {
|
|
"start": {"line": 0, "character": 0},
|
|
"end": _end_position(old_text),
|
|
},
|
|
"text": text,
|
|
}
|
|
]
|
|
else:
|
|
content_changes = [{"text": text}]
|
|
await self._send_notification(
|
|
"textDocument/didChange",
|
|
{
|
|
"textDocument": {"uri": uri, "version": new_version},
|
|
"contentChanges": content_changes,
|
|
},
|
|
)
|
|
# Bumping the version is the whole invalidation story:
|
|
# every stored result tagged with an older version is now
|
|
# stale by definition (see _DocState).
|
|
doc.version = new_version
|
|
doc.text = text
|
|
return new_version
|
|
|
|
# First open: didChangeWatchedFiles CREATED + didOpen.
|
|
await self._send_notification(
|
|
"workspace/didChangeWatchedFiles",
|
|
{"changes": [{"uri": uri, "type": 1}]}, # 1 = CREATED
|
|
)
|
|
# Fresh doc state — anything stashed under this path by a
|
|
# pre-open push (relatedDocuments spillover etc.) is discarded.
|
|
self._docs[abs_path] = _DocState(version=0, text=text)
|
|
await self._send_notification(
|
|
"textDocument/didOpen",
|
|
{
|
|
"textDocument": {
|
|
"uri": uri,
|
|
"languageId": language_id,
|
|
"version": 0,
|
|
"text": text,
|
|
}
|
|
},
|
|
)
|
|
return 0
|
|
|
|
async def save_file(self, path: str) -> None:
|
|
"""Send didSave for ``path``. Some linters re-scan only on save."""
|
|
if not self.is_running:
|
|
return
|
|
abs_path = os.path.abspath(path)
|
|
await self._send_notification(
|
|
"textDocument/didSave",
|
|
{"textDocument": {"uri": file_uri(abs_path)}},
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# diagnostics: pull + wait
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _pull_document_diagnostics(self, path: str) -> None:
|
|
"""Send ``textDocument/diagnostic`` for one file.
|
|
|
|
Stores results into the doc's pull store, tagged with the
|
|
document version captured at request send time. If a didChange
|
|
races past the in-flight request, the version bump makes the
|
|
stored result stale automatically — no explicit invalidation.
|
|
Silently no-ops on errors (server may not support the pull
|
|
endpoint).
|
|
"""
|
|
abs_path = os.path.abspath(path)
|
|
doc = self._docs.get(abs_path)
|
|
sent_version = doc.version if doc else -1
|
|
try:
|
|
params: Dict[str, Any] = {
|
|
"textDocument": {"uri": file_uri(abs_path)}
|
|
}
|
|
result = await self._send_request_with_retry(
|
|
"textDocument/diagnostic",
|
|
params,
|
|
timeout=DIAGNOSTICS_REQUEST_TIMEOUT,
|
|
)
|
|
except (LSPRequestError, LSPProtocolError, asyncio.TimeoutError) as e:
|
|
logger.debug("[%s] document diagnostic pull failed: %s", self.server_id, e)
|
|
return
|
|
if not isinstance(result, dict):
|
|
return
|
|
items = result.get("items")
|
|
if isinstance(items, list):
|
|
doc = self._docs.setdefault(abs_path, _DocState(version=-1))
|
|
doc.pull = items
|
|
doc.pull_version = sent_version
|
|
related = result.get("relatedDocuments")
|
|
if isinstance(related, dict):
|
|
for uri, sub in related.items():
|
|
if not isinstance(sub, dict):
|
|
continue
|
|
sub_items = sub.get("items")
|
|
if isinstance(sub_items, list):
|
|
rel = self._docs.setdefault(uri_to_path(uri), _DocState(version=-1))
|
|
rel.pull = sub_items
|
|
# Same send-anchored tagging: fresh only if that
|
|
# doc hasn't changed since the request went out.
|
|
rel.pull_version = rel.version
|
|
|
|
async def wait_for_diagnostics(
|
|
self,
|
|
path: str,
|
|
version: int,
|
|
*,
|
|
mode: str = "document",
|
|
timeout: Optional[float] = None,
|
|
) -> bool:
|
|
"""Wait for the server to publish diagnostics for ``path`` at ``version``.
|
|
|
|
``mode`` is ``"document"`` (5s budget, document pulls) or
|
|
``"full"`` (10s budget, also workspace pulls). ``timeout``
|
|
overrides the mode's default budget when provided — this is
|
|
how the user's ``lsp.wait_timeout`` config reaches the wait
|
|
loop (slow servers like tsserver on big projects need more
|
|
than the 5s default).
|
|
|
|
Returns ``True`` when *fresh* diagnostics arrived (a push at
|
|
or after our didChange, or a pull answered after it) and
|
|
``False`` on timeout. Callers must treat ``False`` as "no
|
|
data", NOT as "no errors" — the diagnostic stores may still
|
|
hold stale entries from the previous edit at that point.
|
|
Best-effort — never throws if the server doesn't support pull
|
|
diagnostics; we still get the push side.
|
|
"""
|
|
if timeout is not None and timeout > 0:
|
|
budget = timeout
|
|
else:
|
|
budget = DIAGNOSTICS_FULL_WAIT if mode == "full" else DIAGNOSTICS_DOCUMENT_WAIT
|
|
deadline = asyncio.get_event_loop().time() + budget
|
|
abs_path = os.path.abspath(path)
|
|
|
|
while True:
|
|
remaining = deadline - asyncio.get_event_loop().time()
|
|
if remaining <= 0:
|
|
return False
|
|
|
|
# Concurrent: document pull + push wait.
|
|
pull_task = asyncio.create_task(self._pull_document_diagnostics(abs_path))
|
|
push_task = asyncio.create_task(self._wait_for_fresh_push(abs_path, version, remaining))
|
|
done, pending = await asyncio.wait(
|
|
{pull_task, push_task},
|
|
timeout=remaining,
|
|
return_when=asyncio.FIRST_COMPLETED,
|
|
)
|
|
for t in pending:
|
|
t.cancel()
|
|
for t in pending:
|
|
try:
|
|
await t
|
|
except (asyncio.CancelledError, Exception): # noqa: BLE001
|
|
pass
|
|
|
|
# If we got a fresh push for our version, we're done.
|
|
doc = self._docs.get(abs_path)
|
|
if doc and doc.fresh_push(version):
|
|
return True
|
|
|
|
# Pull may have answered for the current version — that's
|
|
# also success.
|
|
if doc and doc.fresh_pull(version):
|
|
return True
|
|
|
|
# Loop until budget runs out.
|
|
|
|
async def _wait_for_fresh_push(self, path: str, version: int, timeout: float) -> None:
|
|
"""Wait until a fresh publishDiagnostics arrives for ``path`` at ``version``+."""
|
|
deadline = asyncio.get_event_loop().time() + timeout
|
|
baseline = self._push_counter
|
|
while True:
|
|
doc = self._docs.get(path)
|
|
if doc and doc.fresh_push(version):
|
|
# Debounce — wait a tick in case more diagnostics arrive
|
|
# immediately after. TS often emits in pairs. We
|
|
# snapshot the counter so we wake on a *new* push, not
|
|
# on the one that satisfied us a moment ago.
|
|
debounce_baseline = self._push_counter
|
|
debounce_deadline = asyncio.get_event_loop().time() + PUSH_DEBOUNCE
|
|
while self._push_counter == debounce_baseline:
|
|
remaining = debounce_deadline - asyncio.get_event_loop().time()
|
|
if remaining <= 0:
|
|
break
|
|
self._push_event.clear()
|
|
try:
|
|
await asyncio.wait_for(self._push_event.wait(), timeout=remaining)
|
|
except asyncio.TimeoutError:
|
|
break
|
|
return
|
|
remaining = deadline - asyncio.get_event_loop().time()
|
|
if remaining <= 0:
|
|
return
|
|
if self._push_counter > baseline:
|
|
# New event arrived but predicate still false — re-check
|
|
# immediately without waiting again.
|
|
baseline = self._push_counter
|
|
continue
|
|
self._push_event.clear()
|
|
try:
|
|
await asyncio.wait_for(self._push_event.wait(), timeout=min(remaining, 0.5))
|
|
except asyncio.TimeoutError:
|
|
continue
|
|
|
|
def diagnostics_for(self, path: str, *, fresh_only: bool = False) -> List[Dict[str, Any]]:
|
|
"""Return current merged + deduped diagnostics for one file.
|
|
|
|
Diagnostics from push and pull stores are concatenated and
|
|
deduplicated by ``(severity, code, message, range)`` content
|
|
key. Empty list if the server hasn't published anything.
|
|
|
|
With ``fresh_only=True``, a store only contributes when its
|
|
version tag has caught up to the document's current version —
|
|
stale leftovers from the previous edit cycle are excluded.
|
|
This is what report paths should use: after an edit, "stale
|
|
errors" and "no errors" must not be conflated.
|
|
"""
|
|
doc = self._docs.get(os.path.abspath(path))
|
|
if doc is None:
|
|
return []
|
|
if fresh_only:
|
|
return _dedupe(
|
|
doc.push if doc.fresh_push() else [],
|
|
doc.pull if doc.fresh_pull() else [],
|
|
)
|
|
return _dedupe(doc.push, doc.pull)
|
|
|
|
|
|
def _dedupe(*lists: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
seen: Set[str] = set()
|
|
out: List[Dict[str, Any]] = []
|
|
for lst in lists:
|
|
for d in lst:
|
|
if not isinstance(d, dict):
|
|
continue
|
|
key = _diagnostic_key(d)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
out.append(d)
|
|
return out
|
|
|
|
|
|
def _diagnostic_key(d: Dict[str, Any]) -> str:
|
|
"""Content-equality key for a diagnostic.
|
|
|
|
Matches the structural-equality used in claude-code's
|
|
``areDiagnosticsEqual`` — message + severity + source + code +
|
|
range coords. The range is reduced to a tuple to keep the key
|
|
stable across dict orderings.
|
|
"""
|
|
rng = d.get("range") or {}
|
|
start = rng.get("start") or {}
|
|
end = rng.get("end") or {}
|
|
code = d.get("code")
|
|
if code is not None and not isinstance(code, str):
|
|
code = str(code)
|
|
return "\x00".join(
|
|
[
|
|
str(d.get("severity") or 1),
|
|
str(code or ""),
|
|
str(d.get("source") or ""),
|
|
str(d.get("message") or "").strip(),
|
|
f"{start.get('line', 0)}:{start.get('character', 0)}-{end.get('line', 0)}:{end.get('character', 0)}",
|
|
]
|
|
)
|
|
|
|
|
|
__all__ = [
|
|
"LSPClient",
|
|
"file_uri",
|
|
"uri_to_path",
|
|
"INITIALIZE_TIMEOUT",
|
|
"DIAGNOSTICS_DOCUMENT_WAIT",
|
|
"DIAGNOSTICS_FULL_WAIT",
|
|
]
|