From 73365e4184552dfd3b7a0bf81e77b057daf4fa0c Mon Sep 17 00:00:00 2001 From: Jack McCarthy Date: Mon, 5 Oct 2026 07:22:26 -0400 Subject: [PATCH 01/10] Retain LSP ownership through verified cleanup --- servers/lean_runtime.py | 137 +++++- servers/lsp/server.py | 448 ++++++++++++++++---- tests/test_lsp_protocol.py | 672 +++++++++++++++++++++++++++++- tests/test_shared_lean_runtime.py | 428 +++++++++++++++++++ 4 files changed, 1595 insertions(+), 90 deletions(-) diff --git a/servers/lean_runtime.py b/servers/lean_runtime.py index b328c797d..39d702b72 100644 --- a/servers/lean_runtime.py +++ b/servers/lean_runtime.py @@ -43,7 +43,6 @@ LspConfig, LspBusyError, LeanLspSession, - LspProtocolError, format_lsp_diagnostics, ) from servers.repl.core import DEFAULT_REPL_STARTUP_TIMEOUT, format_repl_response @@ -422,9 +421,11 @@ def close(self) -> None: resources = [entry.resource for entry in self._entries.values()] self._entries.clear() self._condition.notify_all() - self._close_many(resources) - if self._sweeper and self._sweeper is not threading.current_thread(): - self._sweeper.join() + try: + self._close_many(resources) + finally: + if self._sweeper and self._sweeper is not threading.current_thread(): + self._sweeper.join() def _acquire( self, @@ -543,11 +544,28 @@ def _acquire( self._condition.wait(timeout=wait_seconds) if resources_to_close: - self._close_many(resources_to_close) - resources_to_close.clear() + try: + self._close_many(resources_to_close) + except BaseException: + with self._condition: + if reserved: + self._creating.discard(root) + self._condition.notify_all() + raise + finally: + resources_to_close.clear() if resources_to_close: - self._close_many(resources_to_close) + try: + self._close_many(resources_to_close) + except BaseException: + with self._condition: + if reserved: + self._creating.discard(root) + self._condition.notify_all() + raise + finally: + resources_to_close.clear() if not reserved: return resource @@ -617,8 +635,22 @@ def _sweep(self, interval: float) -> None: logger.exception("failed to evict idle Lean project resources") def _close_many(self, resources: list[T]) -> None: - for resource in resources: - self._safe_close(resource) + first_error: BaseException | None = None + index = 0 + while index < len(resources): + try: + while index < len(resources): + self._safe_close(resources[index]) + index += 1 + except BaseException as error: + if first_error is None: + first_error = error + elif error is not first_error: + add_note = getattr(first_error, "add_note", None) + if add_note is not None: + add_note(f"another Lean resource close was interrupted by: {error}") + if first_error is not None: + raise first_error.with_traceback(first_error.__traceback__) def _safe_close(self, resource: T) -> None: try: @@ -663,7 +695,25 @@ def default_lsp_factory(project_dir: Path) -> LeanLspSession: timeout=self.config.lsp_timeout, ) ) - session.start() + try: + session.start() + except BaseException as start_error: + if not session.is_clean(): + try: + session.abort() + except BaseException as cleanup_error: + if not isinstance(start_error, Exception): + add_note = getattr(start_error, "add_note", None) + if add_note is not None: + add_note( + "Lean LSP startup cleanup was also interrupted by: " + f"{cleanup_error}" + ) + raise start_error.with_traceback( + start_error.__traceback__ + ) from cleanup_error + raise cleanup_error from start_error + raise return session self.repl_projects = ProjectResourceCache( @@ -746,9 +796,24 @@ def dispatch(self, method: str, params: dict[str, Any]) -> Any: diagnostics = session.get_diagnostics(str(path)) except LspBusyError: raise - except (LspProtocolError, TimeoutError, OSError): - session.abort() + except BaseException as operation_error: self.lsp_projects.invalidate(str(root), session) + try: + session.abort() + except BaseException as cleanup_error: + if cleanup_error is operation_error: + raise + if not isinstance(operation_error, Exception): + add_note = getattr(operation_error, "add_note", None) + if add_note is not None: + add_note( + "Lean LSP cleanup was also interrupted by: " + f"{cleanup_error}" + ) + raise operation_error.with_traceback( + operation_error.__traceback__ + ) from cleanup_error + raise cleanup_error from operation_error raise return format_lsp_diagnostics(diagnostics) if method == "lsp.hover": @@ -769,9 +834,24 @@ def dispatch(self, method: str, params: dict[str, Any]) -> Any: result = session.hover(str(path), line, character) except LspBusyError: raise - except (LspProtocolError, TimeoutError, OSError): - session.abort() + except BaseException as operation_error: self.lsp_projects.invalidate(str(root), session) + try: + session.abort() + except BaseException as cleanup_error: + if cleanup_error is operation_error: + raise + if not isinstance(operation_error, Exception): + add_note = getattr(operation_error, "add_note", None) + if add_note is not None: + add_note( + "Lean LSP cleanup was also interrupted by: " + f"{cleanup_error}" + ) + raise operation_error.with_traceback( + operation_error.__traceback__ + ) from cleanup_error + raise cleanup_error from operation_error raise return result or "No hover information at this position." raise ValueError(f"unknown Lean runtime method: {method}") @@ -792,8 +872,33 @@ def status(self, *, include_projects: bool) -> dict[str, Any]: return result def close(self) -> None: - self.repl_projects.close() - self.lsp_projects.close() + close_error: BaseException | None = None + lsp_close_error: BaseException | None = None + try: + self.repl_projects.close() + except BaseException as error: + close_error = error + finally: + while True: + try: + self.lsp_projects.close() + except BaseException as error: + if lsp_close_error is None: + lsp_close_error = error + elif error is not lsp_close_error: + add_note = getattr(lsp_close_error, "add_note", None) + if add_note is not None: + add_note(f"Lean LSP cleanup was also interrupted by: {error}") + continue + break + if close_error is None: + close_error = lsp_close_error + elif lsp_close_error is not None and lsp_close_error is not close_error: + add_note = getattr(close_error, "add_note", None) + if add_note is not None: + add_note(f"Lean LSP cleanup was also interrupted by: {lsp_close_error}") + if close_error is not None: + raise close_error.with_traceback(close_error.__traceback__) def _acquisition_timeout(self, operation_timeout: float) -> float: """Reserve enough of the RPC deadline for the admitted tool operation.""" diff --git a/servers/lsp/server.py b/servers/lsp/server.py index 35af19abd..c5bac6995 100644 --- a/servers/lsp/server.py +++ b/servers/lsp/server.py @@ -21,12 +21,15 @@ from servers import resolve_lean_project_dir from servers.lean_client import LeanRuntimeClient +from servers.repl.core import _kill_subprocesses logger = getLogger(__name__) DEFAULT_LSP_TIMEOUT = 60 MAX_LSP_HEADER_BYTES = 16 * 1024 MAX_LSP_MESSAGE_BYTES = 16 * 1024 * 1024 +LSP_CLEANUP_RETRY_INITIAL_SECONDS = 0.01 +LSP_CLEANUP_RETRY_MAX_SECONDS = 1.0 class LspProtocolError(RuntimeError): @@ -37,6 +40,10 @@ class LspBusyError(TimeoutError): """A queued operation could not enter the shared LSP session in time.""" +class LspCleanupError(LspProtocolError): + """The owned language-server process group could not be retired.""" + + @dataclass class LspConfig: """Configuration for the Lean LSP server.""" @@ -52,8 +59,14 @@ class LeanLspSession: def __init__(self, config: LspConfig) -> None: self.config = config self.process: subprocess.Popen | None = None + self._process_group_id: int | None = None + self._group_retired = True + self._poisoned = False + self._retire_pending = False + self._pending_cancellation: BaseException | None = None self._request_id = 0 self._lock = threading.Lock() + self._lifecycle_lock = threading.Lock() # A session has one stdout stream. Serialize the complete document # lifecycle so concurrent MCP calls cannot race two readers against # that stream or consume one another's diagnostics/responses. @@ -61,67 +74,313 @@ def __init__(self, config: LspConfig) -> None: def start(self) -> None: """Start the language server process.""" - env = os.environ.copy() - env.pop("PYTHONPATH", None) - - self.process = subprocess.Popen( - self.config.lake_command, - cwd=self.config.cwd, - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - env=env, - ) - + if os.name != "posix": + raise RuntimeError("Lean LSP transport requires a POSIX platform") + spawned_process: subprocess.Popen | None = None try: - # An initialize response must be an InitializeResult object. A - # timeout, JSON-RPC error, or malformed result means the backing - # Lean server is unusable, so do not expose an apparently healthy - # MCP server on top of it. - result = self._send_request("initialize", { - "processId": os.getpid(), - "capabilities": {}, - "rootUri": Path(self.config.cwd).resolve().as_uri(), - }) - if not isinstance(result, dict): - raise LspProtocolError( - "LSP initialize returned a non-object result: " - f"{result!r}" - ) + with self._lifecycle_lock: + if not self.is_clean(): + raise RuntimeError("Lean LSP session already owns a process") + env = os.environ.copy() + env.pop("PYTHONPATH", None) + try: + spawned_process = subprocess.Popen( + self.config.lake_command, + cwd=self.config.cwd, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + env=env, + start_new_session=True, + ) + self.process = spawned_process + self._group_retired = False + self._poisoned = True + self._retire_pending = True + self._publish_process_group(spawned_process) + + # An initialize response must be an InitializeResult object. A + # timeout, JSON-RPC error, or malformed result means the backing + # Lean server is unusable, so do not expose an apparently healthy + # MCP server on top of it. + result = self._send_request("initialize", { + "processId": os.getpid(), + "capabilities": {}, + "rootUri": Path(self.config.cwd).resolve().as_uri(), + }) + if not isinstance(result, dict): + raise LspProtocolError( + "LSP initialize returned a non-object result: " + f"{result!r}" + ) + + self._send_notification("initialized", {}) + except BaseException as start_error: + if self.process is None and spawned_process is not None: + self.process = spawned_process + if spawned_process is not None and self._process_group_id is None: + self._group_retired = False + self._poisoned = True + self._retire_pending = True + self._remember_cancellation(start_error) + try: + self._retire_until_clean("after startup failure") + except BaseException as cleanup_error: + if cleanup_error is start_error: + raise + raise cleanup_error from start_error + raise + except BaseException as start_error: + if self.is_clean(): + raise + if self.process is None and spawned_process is not None: + self.process = spawned_process + self._group_retired = False + self._poisoned = True + self._retire_pending = True + self._recover_interrupted_retirement( + start_error, + "after interrupted startup", + ) - self._send_notification("initialized", {}) - except BaseException: - self._abort_process() - raise + def _publish_process_group(self, process: subprocess.Popen) -> None: + """Publish complete ownership immediately after a successful spawn.""" + self._process_group_id = process.pid + self._group_retired = False + self._poisoned = False + self._retire_pending = False def _abort_process(self) -> None: - """Force-close the backing process without attempting more JSON-RPC.""" - process, self.process = self.process, None - if process is None or process.poll() is not None: + """Force-close the complete owned process group and verify cleanup.""" + if self.is_clean(): + self._raise_pending_cancellation() return - try: - process.kill() - process.wait(timeout=5) - except Exception: - pass + self._poisoned = True + self._retire_pending = True + self._complete_retirement("after abort") def close(self) -> None: """Shut down the language server.""" - if self.process and self.process.poll() is None: - try: - self._send_request("shutdown", {}) - self._send_notification("exit", {}) - self.process.wait(timeout=5) - except Exception: - self._abort_process() - self.process = None + try: + with self._lifecycle_lock: + try: + self._poisoned = True + if self.is_clean(): + self._raise_pending_cancellation() + return + process = self.process + already_retiring = self._retire_pending + self._retire_pending = True + if not already_retiring and process is not None: + if process.poll() is None: + self._send_request("shutdown", {}) + self._send_notification("exit", {}) + process.wait(timeout=5) + except BaseException as error: + self._remember_cancellation(error) + self._poisoned = True + if not self.is_clean(): + self._retire_pending = True + if isinstance(error, Exception): + logger.warning( + "graceful Lean LSP shutdown failed; enforcing cleanup", + exc_info=True, + ) + self._complete_retirement("during close") + except BaseException as error: + if self.is_clean(): + raise + self._recover_interrupted_retirement(error, "during interrupted close") def abort(self) -> None: """Discard a protocol stream that can no longer be shared safely.""" - self._abort_process() + try: + with self._lifecycle_lock: + try: + self._poisoned = True + self._abort_process() + except BaseException as error: + if self.is_clean(): + raise + self._remember_cancellation(error) + self._poisoned = True + self._retire_pending = True + self._complete_retirement("after interrupted abort") + except BaseException as error: + if self.is_clean(): + raise + self._recover_interrupted_retirement(error, "after interrupted abort") def is_alive(self) -> bool: """Return whether the cached language-server child can accept work.""" - return self.process is not None and self.process.poll() is None + return ( + self.process is not None + and not self._poisoned + and not self._retire_pending + and self.process.poll() is None + ) + + def is_clean(self) -> bool: + """Return whether this object owns no process, group, or open pipes.""" + return ( + self.process is None + and self._process_group_id is None + and not self._retire_pending + ) + + def _mark_retire_pending(self) -> None: + """Poison the protocol stream before another admitted operation sees it.""" + self._poisoned = True + self._retire_pending = True + + def _require_usable_after_admission(self) -> None: + if self._poisoned or self._retire_pending: + raise LspProtocolError("Lean LSP session is retiring after a failed operation") + + def _remember_cancellation(self, error: BaseException) -> None: + if isinstance(error, Exception): + return + if self._pending_cancellation is None: + self._pending_cancellation = error + return + add_note = getattr(self._pending_cancellation, "add_note", None) + if add_note is not None and error is not self._pending_cancellation: + add_note(f"Lean LSP cleanup was also interrupted by: {error}") + + def _raise_pending_cancellation(self) -> None: + error, self._pending_cancellation = self._pending_cancellation, None + if error is not None: + raise error.with_traceback(error.__traceback__) + + @staticmethod + def _close_process_streams(process: subprocess.Popen) -> None: + first_error: Exception | None = None + first_cancellation: BaseException | None = None + for name in ("stdin", "stdout", "stderr"): + stream = getattr(process, name, None) + if stream is None: + continue + try: + stream.close() + except BaseException as error: + if isinstance(error, Exception): + if first_error is None: + first_error = error + elif first_cancellation is None: + first_cancellation = error + else: + add_note = getattr(first_cancellation, "add_note", None) + if add_note is not None: + add_note(f"Lean LSP pipe cleanup was also interrupted by: {error}") + else: + setattr(process, name, None) + if first_cancellation is not None: + if first_error is not None: + add_note = getattr(first_cancellation, "add_note", None) + if add_note is not None: + add_note(f"Lean LSP pipe cleanup also failed: {first_error}") + raise first_cancellation + if first_error is not None: + raise LspCleanupError("failed to close Lean LSP process pipes") from first_error + + def _retire_owned_process_once(self) -> None: + process = self.process + process_group_id = self._process_group_id + if process is None and process_group_id is None: + self._retire_pending = False + return + if process is None: + raise LspCleanupError("Lean LSP process ownership is incomplete") + if not self._group_retired: + if process_group_id is None: + try: + process_group_id = process.pid + except Exception as error: + raise LspCleanupError( + "Lean LSP process-group id is unavailable" + ) from error + self._process_group_id = process_group_id + try: + assert process_group_id is not None + _kill_subprocesses(process, process_group_id) + except Exception as error: + raise LspCleanupError( + "failed to retire the Lean LSP process group" + ) from error + # Publish verified retirement before clearing the PGID. If a + # cancellation lands between these stores, the retry sees this + # flag and cannot signal a subsequently reused process-group id. + self._group_retired = True + self._process_group_id = None + self._close_process_streams(process) + if all(getattr(process, name, None) is None for name in ("stdin", "stdout", "stderr")): + self.process = None + self._retire_pending = False + + def _retire_owned_process(self) -> None: + """Retry cancellation-like interruptions, but expose cleanup failures.""" + while not self.is_clean(): + try: + self._retire_owned_process_once() + except Exception: + raise + except BaseException as error: + self._remember_cancellation(error) + self._raise_pending_cancellation() + + def _retire_until_clean(self, context: str) -> None: + """Retain ownership and retry until every child and pipe is clean.""" + delay = LSP_CLEANUP_RETRY_INITIAL_SECONDS + while not self.is_clean(): + try: + self._retire_owned_process() + except LspCleanupError: + logger.exception("Lean LSP cleanup failed %s; retrying", context) + except BaseException: + # `_retire_owned_process` releases a remembered cancellation only + # after cleanup is verified. + if self.is_clean(): + raise + raise + if self.is_clean(): + break + try: + time.sleep(delay) + except BaseException as error: + self._remember_cancellation(error) + delay = min(delay * 2, LSP_CLEANUP_RETRY_MAX_SECONDS) + self._raise_pending_cancellation() + + def _complete_retirement(self, context: str) -> None: + """Do not let an interruption escape while this object still owns state.""" + while not self.is_clean(): + try: + self._retire_until_clean(context) + except BaseException as error: + if self.is_clean(): + raise + self._remember_cancellation(error) + if isinstance(error, Exception): + logger.exception("Lean LSP retirement failed %s; retrying", context) + self._raise_pending_cancellation() + + def _recover_interrupted_retirement( + self, + error: BaseException, + context: str, + ) -> None: + """Reacquire lifecycle ownership until cancellation-safe cleanup finishes.""" + self._remember_cancellation(error) + while not self.is_clean(): + try: + with self._lifecycle_lock: + self._poisoned = True + self._retire_pending = True + self._complete_retirement(context) + except BaseException as retry_error: + if self.is_clean(): + raise + self._remember_cancellation(retry_error) + self._raise_pending_cancellation() def get_diagnostics(self, file_path: str) -> list[dict]: """Open a file and collect diagnostics from the language server.""" @@ -131,12 +390,17 @@ def get_diagnostics(self, file_path: str) -> list[dict]: f"timed out after {self.config.timeout:g}s waiting for the Lean LSP session" ) try: + self._require_usable_after_admission() remaining = deadline - time.monotonic() if remaining <= 0: raise LspBusyError( f"timed out after {self.config.timeout:g}s waiting for the Lean LSP session" ) - return self._get_diagnostics(file_path, timeout=remaining) + try: + return self._get_diagnostics(file_path, timeout=remaining) + except BaseException: + self._mark_retire_pending() + raise finally: self._operation_lock.release() @@ -164,25 +428,25 @@ def _get_diagnostics(self, file_path: str, *, timeout: float | None = None) -> l timeout=self._remaining(deadline, operation_timeout), ) + diagnostics: list[dict] | None = None + operation_error: BaseException | None = None try: # An empty published diagnostic list means the file is clean. No # publication at all is a timeout/error and must never be conflated # with that valid empty result. - return self._collect_diagnostics( + diagnostics = self._collect_diagnostics( uri, timeout=self._remaining(deadline, operation_timeout), ) - finally: - try: - remaining = deadline - time.monotonic() - if remaining > 0: - self._send_notification( - "textDocument/didClose", - {"textDocument": {"uri": uri}}, - timeout=remaining, - ) - except Exception: - logger.warning("failed to close LSP document %s", uri, exc_info=True) + except BaseException as error: + operation_error = error + close_error = self._close_document(uri, deadline) + if operation_error is not None: + self._raise_operation_error(operation_error, close_error) + if close_error is not None: + raise close_error + assert diagnostics is not None + return diagnostics def hover(self, file_path: str, line: int, character: int) -> str | None: """Get hover information at a position.""" @@ -192,12 +456,17 @@ def hover(self, file_path: str, line: int, character: int) -> str | None: f"timed out after {self.config.timeout:g}s waiting for the Lean LSP session" ) try: + self._require_usable_after_admission() remaining = deadline - time.monotonic() if remaining <= 0: raise LspBusyError( f"timed out after {self.config.timeout:g}s waiting for the Lean LSP session" ) - return self._hover(file_path, line, character, timeout=remaining) + try: + return self._hover(file_path, line, character, timeout=remaining) + except BaseException: + self._mark_retire_pending() + raise finally: self._operation_lock.release() @@ -225,6 +494,8 @@ def _hover( }, timeout=self._remaining(deadline, timeout), ) + result: Any = None + operation_error: BaseException | None = None try: result = self._send_request( "textDocument/hover", @@ -234,17 +505,13 @@ def _hover( }, timeout=self._remaining(deadline, timeout), ) - finally: - try: - remaining = deadline - time.monotonic() - if remaining > 0: - self._send_notification( - "textDocument/didClose", - {"textDocument": {"uri": uri}}, - timeout=remaining, - ) - except Exception: - logger.warning("failed to close LSP document %s", uri, exc_info=True) + except BaseException as error: + operation_error = error + close_error = self._close_document(uri, deadline) + if operation_error is not None: + self._raise_operation_error(operation_error, close_error) + if close_error is not None: + raise close_error if result and "contents" in result: contents = result["contents"] if isinstance(contents, dict): @@ -252,6 +519,43 @@ def _hover( return str(contents) return None + def _close_document(self, uri: str, deadline: float) -> BaseException | None: + remaining = deadline - time.monotonic() + if remaining <= 0: + return TimeoutError( + "Lean LSP operation deadline expired before closing the document" + ) + try: + self._send_notification( + "textDocument/didClose", + {"textDocument": {"uri": uri}}, + timeout=remaining, + ) + except BaseException as error: + return error + return None + + @staticmethod + def _raise_operation_error( + operation_error: BaseException, + close_error: BaseException | None, + ) -> None: + if close_error is not None and close_error is not operation_error: + if isinstance(operation_error, Exception) and not isinstance( + close_error, Exception + ): + add_note = getattr(close_error, "add_note", None) + if add_note is not None: + add_note(f"the Lean LSP operation also failed: {operation_error}") + raise close_error.with_traceback(close_error.__traceback__) from operation_error + add_note = getattr(operation_error, "add_note", None) + if add_note is not None: + add_note(f"closing the Lean LSP document also failed: {close_error}") + raise operation_error.with_traceback( + operation_error.__traceback__ + ) from close_error + raise operation_error.with_traceback(operation_error.__traceback__) + def _send_request(self, method: str, params: dict, *, timeout: float = 30) -> Any: """Send a JSON-RPC request and wait for response.""" deadline = time.monotonic() + timeout diff --git a/tests/test_lsp_protocol.py b/tests/test_lsp_protocol.py index eedcf5b84..39edf1b4b 100644 --- a/tests/test_lsp_protocol.py +++ b/tests/test_lsp_protocol.py @@ -3,6 +3,9 @@ from __future__ import annotations import os +import signal +import subprocess +import sys import threading from pathlib import Path from types import SimpleNamespace @@ -13,10 +16,14 @@ class _FakeProcess: - def __init__(self) -> None: + def __init__(self, *, pid: int = 43210) -> None: + self.pid = pid self.returncode: int | None = None self.killed = False self.waited = False + self.stdin = _FakeStream() + self.stdout = _FakeStream() + self.stderr = None def poll(self): return self.returncode @@ -30,6 +37,42 @@ def wait(self, timeout=None): return self.returncode +class _FakeStream: + def __init__( + self, + *, + failures: int = 0, + errors: list[BaseException] | None = None, + events: list[str] | None = None, + name: str = "", + ) -> None: + self.failures = failures + self.errors = list(errors or []) + self.events = events + self.name = name + self.close_calls = 0 + + def close(self) -> None: + self.close_calls += 1 + if self.events is not None: + self.events.append(f"close:{self.name}") + if self.failures: + self.failures -= 1 + raise OSError(f"cannot close {self.name}") + if self.errors: + raise self.errors.pop(0) + + +class _Cancellation(BaseException): + pass + + +def _finish_fake_process(process: _FakeProcess) -> None: + process.killed = True + process.returncode = -9 + process.waited = True + + def test_json_rpc_error_response_raises_protocol_error(monkeypatch): session = lsp.LeanLspSession(lsp.LspConfig()) monkeypatch.setattr( @@ -48,7 +91,18 @@ def test_json_rpc_error_response_raises_protocol_error(monkeypatch): def test_start_aborts_process_when_initialize_fails(monkeypatch): process = _FakeProcess() - monkeypatch.setattr(lsp.subprocess, "Popen", lambda *args, **kwargs: process) + popen_kwargs = {} + + def popen(*args, **kwargs): + popen_kwargs.update(kwargs) + return process + + monkeypatch.setattr(lsp.subprocess, "Popen", popen) + monkeypatch.setattr( + lsp, + "_kill_subprocesses", + lambda owned, process_group_id: _finish_fake_process(owned), + ) session = lsp.LeanLspSession(lsp.LspConfig()) @@ -62,7 +116,621 @@ def fail_initialize(method, params): assert process.killed is True assert process.waited is True + assert popen_kwargs["start_new_session"] is True assert session.process is None + assert session._process_group_id is None + assert session.is_clean() + + +def test_start_rejects_an_unsupported_platform_before_spawning(monkeypatch): + monkeypatch.setattr(lsp.os, "name", "nt") + monkeypatch.setattr( + lsp.subprocess, + "Popen", + lambda *args, **kwargs: pytest.fail("unsupported LSP start spawned a process"), + ) + + with pytest.raises(RuntimeError, match="POSIX"): + lsp.LeanLspSession(lsp.LspConfig()).start() + + +def test_startup_retries_cleanup_without_losing_the_initialization_error(monkeypatch): + process = _FakeProcess() + attempts = 0 + monkeypatch.setattr(lsp.subprocess, "Popen", lambda *args, **kwargs: process) + monkeypatch.setattr(lsp.time, "sleep", lambda delay: None) + + def retire(owned, process_group_id): + nonlocal attempts + attempts += 1 + if attempts == 1: + raise RuntimeError("stubborn process group") + _finish_fake_process(owned) + + monkeypatch.setattr(lsp, "_kill_subprocesses", retire) + session = lsp.LeanLspSession(lsp.LspConfig()) + monkeypatch.setattr( + session, + "_send_request", + lambda method, params: (_ for _ in ()).throw( + lsp.LspProtocolError("initialize rejected") + ), + ) + + with pytest.raises(lsp.LspProtocolError, match="initialize rejected"): + session.start() + + assert attempts == 2 + assert session.is_clean() + + +def test_startup_publication_cancellation_still_retires_the_uncached_child(monkeypatch): + process = _FakeProcess() + session = lsp.LeanLspSession(lsp.LspConfig()) + monkeypatch.setattr(lsp.subprocess, "Popen", lambda *args, **kwargs: process) + monkeypatch.setattr( + session, + "_publish_process_group", + lambda owned: (_ for _ in ()).throw( + _Cancellation("cancel process-group publication") + ), + ) + retired = [] + + def retire(owned, process_group_id): + retired.append((owned, process_group_id)) + _finish_fake_process(owned) + + monkeypatch.setattr(lsp, "_kill_subprocesses", retire) + + with pytest.raises(_Cancellation, match="cancel process-group publication"): + session.start() + + assert retired == [(process, process.pid)] + assert session.is_clean() + + +def test_startup_lock_exit_cancellation_retires_the_initialized_child(monkeypatch): + process = _FakeProcess() + session = lsp.LeanLspSession(lsp.LspConfig()) + inner_lock = threading.Lock() + + class InterruptingLock: + def __init__(self): + self.interrupt = True + + def __enter__(self): + inner_lock.acquire() + return self + + def __exit__(self, exc_type, exc_value, traceback): + inner_lock.release() + if self.interrupt: + self.interrupt = False + raise _Cancellation("cancel lifecycle-lock exit") + return False + + session._lifecycle_lock = InterruptingLock() + monkeypatch.setattr(lsp.subprocess, "Popen", lambda *args, **kwargs: process) + monkeypatch.setattr(session, "_send_request", lambda method, params: {}) + monkeypatch.setattr(session, "_send_notification", lambda method, params: None) + retired = [] + + def retire(owned, process_group_id): + retired.append((owned, process_group_id)) + _finish_fake_process(owned) + + monkeypatch.setattr(lsp, "_kill_subprocesses", retire) + + with pytest.raises(_Cancellation, match="cancel lifecycle-lock exit"): + session.start() + + assert retired == [(process, process.pid)] + assert session.is_clean() + + +def test_close_preserves_cancellation_until_process_and_pipes_are_clean(monkeypatch): + process = _FakeProcess() + stdin = process.stdin + stdout = process.stdout + session = lsp.LeanLspSession(lsp.LspConfig()) + session.process = process + session._process_group_id = process.pid + session._group_retired = False + events = [] + + def cancel_shutdown(method, params): + events.append("cancel") + raise _Cancellation("cancel shutdown") + + def retire(owned, process_group_id): + events.append("retire") + _finish_fake_process(owned) + + monkeypatch.setattr(session, "_send_request", cancel_shutdown) + monkeypatch.setattr(lsp, "_kill_subprocesses", retire) + + with pytest.raises(_Cancellation, match="cancel shutdown"): + session.close() + + assert events == ["cancel", "retire"] + assert stdin.close_calls == 1 + assert stdout.close_calls == 1 + assert session.is_clean() + + +def test_cleanup_retry_retains_ownership_and_never_resignals_after_group_exit(monkeypatch): + events: list[str] = [] + process = _FakeProcess() + process.stdin = _FakeStream(failures=1, events=events, name="stdin") + process.stdout = _FakeStream(events=events, name="stdout") + session = lsp.LeanLspSession(lsp.LspConfig()) + session.process = process + session._process_group_id = process.pid + session._group_retired = False + session._retire_pending = True + retire_calls = 0 + + def retire(owned, process_group_id): + nonlocal retire_calls + retire_calls += 1 + events.append("retire") + _finish_fake_process(owned) + + monkeypatch.setattr(lsp, "_kill_subprocesses", retire) + monkeypatch.setattr(lsp.time, "sleep", lambda delay: events.append("retry")) + + session.abort() + + assert events == ["retire", "close:stdin", "close:stdout", "retry", "close:stdin"] + assert retire_calls == 1 + assert process.stdin is None + assert process.stdout is None + assert session.is_clean() + + +def test_cancellation_between_group_retirement_stores_never_resignals(monkeypatch): + class Session(lsp.LeanLspSession): + def __setattr__(self, name, value): + if ( + name == "_group_retired" + and value is True + and getattr(self, "interrupt_retirement_store", False) + ): + object.__setattr__(self, name, value) + object.__setattr__(self, "interrupt_retirement_store", False) + raise _Cancellation("cancel retirement publication") + super().__setattr__(name, value) + + process = _FakeProcess() + session = Session(lsp.LspConfig()) + session.process = process + session._process_group_id = process.pid + session._group_retired = False + session._retire_pending = True + session.interrupt_retirement_store = True + retire_calls = 0 + + def retire(owned, process_group_id): + nonlocal retire_calls + retire_calls += 1 + _finish_fake_process(owned) + + monkeypatch.setattr(lsp, "_kill_subprocesses", retire) + + with pytest.raises(_Cancellation, match="cancel retirement publication"): + session.abort() + + assert retire_calls == 1 + assert session.is_clean() + + +def test_pipe_cleanup_preserves_cancellation_over_an_ordinary_close_error(monkeypatch): + events: list[str] = [] + cancellation = _Cancellation("cancel pipe cleanup") + process = _FakeProcess() + process.stdin = _FakeStream(failures=1, events=events, name="stdin") + process.stdout = _FakeStream( + errors=[cancellation], + events=events, + name="stdout", + ) + stdin = process.stdin + stdout = process.stdout + session = lsp.LeanLspSession(lsp.LspConfig()) + session.process = process + session._process_group_id = process.pid + session._group_retired = False + session._retire_pending = True + retire_calls = 0 + + def retire(owned, process_group_id): + nonlocal retire_calls + retire_calls += 1 + events.append("retire") + _finish_fake_process(owned) + + monkeypatch.setattr(lsp, "_kill_subprocesses", retire) + + with pytest.raises(_Cancellation, match="cancel pipe cleanup"): + session.abort() + + assert retire_calls == 1 + assert session.is_clean() + assert process.stdin is None + assert process.stdout is None + assert stdin.close_calls == 2 + assert stdout.close_calls == 2 + + +def test_cleanup_retry_preserves_sleep_cancellation_until_cleanup_succeeds(monkeypatch): + process = _FakeProcess() + session = lsp.LeanLspSession(lsp.LspConfig()) + session.process = process + session._process_group_id = process.pid + session._group_retired = False + session._retire_pending = True + attempts = 0 + sleeps = 0 + + def retire(owned, process_group_id): + nonlocal attempts + attempts += 1 + if attempts == 1: + raise RuntimeError("retry cleanup") + _finish_fake_process(owned) + + def interrupt_sleep(delay): + nonlocal sleeps + sleeps += 1 + raise _Cancellation("cancel retry sleep") + + monkeypatch.setattr(lsp, "_kill_subprocesses", retire) + monkeypatch.setattr(lsp.time, "sleep", interrupt_sleep) + + with pytest.raises(_Cancellation, match="cancel retry sleep"): + session.abort() + + assert attempts == 2 + assert sleeps == 1 + assert session.is_clean() + + +@pytest.mark.parametrize("operation", ["abort", "close"]) +def test_retirement_transition_cannot_be_cancelled_before_cleanup( + monkeypatch, operation +): + process = _FakeProcess() + session = lsp.LeanLspSession(lsp.LspConfig()) + session.process = process + session._process_group_id = process.pid + session._group_retired = False + session._retire_pending = True + original_retire = session._retire_until_clean + calls = 0 + + def interrupt_before_cleanup(context): + nonlocal calls + calls += 1 + if calls == 1: + raise _Cancellation("cancel before cleanup") + return original_retire(context) + + monkeypatch.setattr(session, "_retire_until_clean", interrupt_before_cleanup) + monkeypatch.setattr( + lsp, + "_kill_subprocesses", + lambda owned, process_group_id: _finish_fake_process(owned), + ) + + with pytest.raises(_Cancellation, match="cancel before cleanup"): + getattr(session, operation)() + + assert calls == 2 + assert session.is_clean() + + +def test_failed_operation_poisons_a_queued_waiter_before_releasing_admission(monkeypatch): + session = lsp.LeanLspSession(lsp.LspConfig()) + first_entered = threading.Event() + release_first = threading.Event() + second_touched_transport = threading.Event() + errors: list[BaseException] = [] + calls = 0 + calls_lock = threading.Lock() + + def diagnostics(file_path, *, timeout=None): + nonlocal calls + with calls_lock: + calls += 1 + number = calls + if number == 1: + first_entered.set() + assert release_first.wait(timeout=2) + raise lsp.LspProtocolError("broken shared stream") + second_touched_transport.set() + return [] + + monkeypatch.setattr(session, "_get_diagnostics", diagnostics) + + def run() -> None: + try: + session.get_diagnostics("ignored.lean") + except BaseException as error: + errors.append(error) + + first = threading.Thread(target=run) + second = threading.Thread(target=run) + first.start() + assert first_entered.wait(timeout=1) + second.start() + release_first.set() + first.join(timeout=2) + second.join(timeout=2) + + assert not first.is_alive() + assert not second.is_alive() + assert not second_touched_transport.is_set() + assert len(errors) == 2 + messages = {str(error) for error in errors} + assert any("broken shared stream" in message for message in messages) + assert any("session is retiring" in message for message in messages) + + +def test_verified_cleanup_does_not_make_a_poisoned_session_reusable(monkeypatch): + session = lsp.LeanLspSession(lsp.LspConfig()) + session._mark_retire_pending() + session.abort() + assert session.is_clean() + assert not session.is_alive() + monkeypatch.setattr( + session, + "_get_diagnostics", + lambda *args, **kwargs: pytest.fail("a cleaned poisoned stream was reused"), + ) + + with pytest.raises(lsp.LspProtocolError, match="session is retiring"): + session.get_diagnostics("ignored.lean") + + +@pytest.mark.parametrize("operation", ["diagnostics", "hover"]) +def test_did_close_failure_poisons_a_queued_operation( + tmp_path, monkeypatch, operation +): + source = tmp_path / "Test.lean" + source.write_text("#check Nat\n") + session = lsp.LeanLspSession(lsp.LspConfig()) + close_entered = threading.Event() + release_close = threading.Event() + notifications = [] + errors: list[BaseException] = [] + + def notify(method, params, **kwargs): + notifications.append(method) + if method == "textDocument/didClose": + close_entered.set() + assert release_close.wait(timeout=2) + raise lsp.LspProtocolError("didClose write failed") + + monkeypatch.setattr(session, "_send_notification", notify) + monkeypatch.setattr(session, "_collect_diagnostics", lambda uri, timeout: []) + monkeypatch.setattr( + session, + "_send_request", + lambda method, params, timeout=30: { + "contents": {"kind": "plaintext", "value": "Nat : Type"} + }, + ) + + def run() -> None: + try: + if operation == "diagnostics": + session.get_diagnostics(str(source)) + else: + session.hover(str(source), 0, 0) + except BaseException as error: + errors.append(error) + + first = threading.Thread(target=run) + second = threading.Thread(target=run) + first.start() + assert close_entered.wait(timeout=1) + second.start() + release_close.set() + first.join(timeout=2) + second.join(timeout=2) + + assert not first.is_alive() + assert not second.is_alive() + assert len(errors) == 2 + assert any("didClose write failed" in str(error) for error in errors) + assert any("session is retiring" in str(error) for error in errors) + assert notifications == ["textDocument/didOpen", "textDocument/didClose"] + + +def test_document_close_budget_exhaustion_poisons_the_session(tmp_path, monkeypatch): + source = tmp_path / "Test.lean" + source.write_text("#check Nat\n") + session = lsp.LeanLspSession(lsp.LspConfig(timeout=60)) + clock = iter([0.0, 0.0, 0.0, 0.0, 60.0]) + notifications = [] + monkeypatch.setattr(lsp.time, "monotonic", lambda: next(clock, 60.0)) + monkeypatch.setattr( + session, + "_send_notification", + lambda method, params, **kwargs: notifications.append(method), + ) + monkeypatch.setattr(session, "_collect_diagnostics", lambda uri, timeout: []) + + with pytest.raises(TimeoutError, match="before closing the document"): + session.get_diagnostics(str(source)) + + assert notifications == ["textDocument/didOpen"] + assert session._poisoned is True + assert session._retire_pending is True + + +@pytest.mark.parametrize("operation", ["diagnostics", "hover"]) +def test_did_close_cancellation_outranks_an_ordinary_operation_error( + tmp_path, monkeypatch, operation +): + source = tmp_path / "Test.lean" + source.write_text("#check Nat\n") + session = lsp.LeanLspSession(lsp.LspConfig()) + operation_error = lsp.LspProtocolError("operation failed") + close_cancellation = _Cancellation("cancel didClose") + + def notify(method, params, **kwargs): + if method == "textDocument/didClose": + raise close_cancellation + + monkeypatch.setattr(session, "_send_notification", notify) + monkeypatch.setattr( + session, + "_collect_diagnostics", + lambda uri, timeout: (_ for _ in ()).throw(operation_error), + ) + monkeypatch.setattr( + session, + "_send_request", + lambda method, params, timeout=30: (_ for _ in ()).throw(operation_error), + ) + + with pytest.raises(_Cancellation, match="cancel didClose") as raised: + if operation == "diagnostics": + session.get_diagnostics(str(source)) + else: + session.hover(str(source), 0, 0) + + assert raised.value is close_cancellation + assert raised.value.__cause__ is operation_error + assert session._poisoned is True + + +@pytest.mark.parametrize("operation", ["diagnostics", "hover"]) +def test_operation_cancellation_outranks_an_ordinary_did_close_error( + tmp_path, monkeypatch, operation +): + source = tmp_path / "Test.lean" + source.write_text("#check Nat\n") + session = lsp.LeanLspSession(lsp.LspConfig()) + operation_cancellation = _Cancellation("cancel operation") + close_error = lsp.LspProtocolError("didClose failed") + + def notify(method, params, **kwargs): + if method == "textDocument/didClose": + raise close_error + + monkeypatch.setattr(session, "_send_notification", notify) + monkeypatch.setattr( + session, + "_collect_diagnostics", + lambda uri, timeout: (_ for _ in ()).throw(operation_cancellation), + ) + monkeypatch.setattr( + session, + "_send_request", + lambda method, params, timeout=30: (_ for _ in ()).throw( + operation_cancellation + ), + ) + + with pytest.raises(_Cancellation, match="cancel operation") as raised: + if operation == "diagnostics": + session.get_diagnostics(str(source)) + else: + session.hover(str(source), 0, 0) + + assert raised.value is operation_cancellation + assert raised.value.__cause__ is close_error + assert session._poisoned is True + + +def test_concurrent_abort_and_close_serialize_owned_cleanup(monkeypatch): + process = _FakeProcess() + session = lsp.LeanLspSession(lsp.LspConfig()) + session.process = process + session._process_group_id = process.pid + session._group_retired = False + session._retire_pending = True + entered = threading.Event() + release = threading.Event() + active = 0 + maximum_active = 0 + errors: list[BaseException] = [] + state_lock = threading.Lock() + + def retire(owned, process_group_id): + nonlocal active, maximum_active + with state_lock: + active += 1 + maximum_active = max(maximum_active, active) + entered.set() + assert release.wait(timeout=2) + _finish_fake_process(owned) + with state_lock: + active -= 1 + + monkeypatch.setattr(lsp, "_kill_subprocesses", retire) + + def call(operation) -> None: + try: + operation() + except BaseException as error: + errors.append(error) + + aborter = threading.Thread(target=call, args=(session.abort,)) + closer = threading.Thread(target=call, args=(session.close,)) + aborter.start() + assert entered.wait(timeout=1) + closer.start() + assert closer.is_alive() + release.set() + aborter.join(timeout=2) + closer.join(timeout=2) + + assert errors == [] + assert maximum_active == 1 + assert session.is_clean() + + +@pytest.mark.skipif(os.name != "posix", reason="native LSP backend requires POSIX groups") +def test_close_retires_a_descendant_after_the_lsp_wrapper_exits(): + script = ( + "import subprocess, sys\n" + "child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(60)'])\n" + "print(child.pid, flush=True)\n" + ) + process = subprocess.Popen( + [sys.executable, "-c", script], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + start_new_session=True, + ) + assert process.stdout is not None + child_pid = int(process.stdout.readline()) + assert process.wait(timeout=5) == 0 + session = lsp.LeanLspSession(lsp.LspConfig()) + session.process = process + session._process_group_id = process.pid + session._group_retired = False + try: + session.close() + assert session.is_clean() + try: + import psutil + + child = psutil.Process(child_pid) + assert child.status() == psutil.STATUS_ZOMBIE + except psutil.NoSuchProcess: + pass + finally: + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass def test_diagnostics_wait_through_initial_quiet_period(monkeypatch): diff --git a/tests/test_shared_lean_runtime.py b/tests/test_shared_lean_runtime.py index 8b61bec55..8e545a374 100644 --- a/tests/test_shared_lean_runtime.py +++ b/tests/test_shared_lean_runtime.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import os import socket import subprocess import sys @@ -657,3 +658,430 @@ def get_diagnostics(self, file_path): assert sessions[0].closed is True finally: services.close() + + +def test_failed_lsp_cleanup_is_invalidated_and_settled_before_replacement( + tmp_path, monkeypatch +): + from servers.lsp import server as lsp + + project = make_lake_project(tmp_path, "lsp-cleanup-order") + source = project / "Main.lean" + source.write_text("#check Nat\n") + sessions = [] + errors = [] + results = [] + events = [] + cleanup_started = threading.Event() + release_cleanup = threading.Event() + attempts = 0 + services = None + + class Session(lsp.LeanLspSession): + def __init__(self, root): + super().__init__(lsp.LspConfig(cwd=str(root))) + self.number = len(sessions) + 1 + self.process = _RuntimeFakeProcess(50000 + self.number) + self._process_group_id = self.process.pid + self._group_retired = False + sessions.append(self) + events.append(f"create:{self.number}") + + def get_diagnostics(self, file_path): + if self.number == 1: + raise lsp.LspProtocolError("broken shared stream") + return [] + + def close(self): + self.abort() + + class _RuntimeFakeProcess: + def __init__(self, pid): + self.pid = pid + self.returncode = None + self.stdin = None + self.stdout = None + self.stderr = None + + def poll(self): + return self.returncode + + def wait(self, timeout=None): + return self.returncode + + def kill(self): + self.returncode = -9 + + def retire(process, process_group_id): + nonlocal attempts + if process is sessions[0].process: + attempts += 1 + resident = services.lsp_projects.stats()["resident"] + events.append(f"retire:{attempts}:valid={resident[0]['valid']}") + if attempts == 1: + cleanup_started.set() + raise RuntimeError("transient cleanup failure") + assert release_cleanup.wait(timeout=2) + process.returncode = -9 + + monkeypatch.setattr(lsp, "_kill_subprocesses", retire) + monkeypatch.setattr(lsp.time, "sleep", lambda delay: None) + services = LeanRuntimeServices( + runtime_config(), + repl_factory=FakePool, + lsp_factory=Session, + start_sweepers=False, + ) + params = {"project_dir": str(project), "file_path": "Main.lean"} + + def dispatch(target): + try: + target.append(services.dispatch("lsp.diagnostics", params)) + except BaseException as error: + errors.append(error) + + first = threading.Thread(target=dispatch, args=([],)) + second = threading.Thread(target=dispatch, args=(results,)) + try: + first.start() + assert cleanup_started.wait(timeout=1) + second.start() + assert len(sessions) == 1 + assert not results + + release_cleanup.set() + first.join(timeout=2) + second.join(timeout=2) + + assert not first.is_alive() + assert not second.is_alive() + assert len(errors) == 1 + assert isinstance(errors[0], lsp.LspProtocolError) + assert str(errors[0]) == "broken shared stream" + assert results == ["No diagnostics — file compiles cleanly."] + assert events[:4] == [ + "create:1", + "retire:1:valid=False", + "retire:2:valid=False", + "create:2", + ] + finally: + release_cleanup.set() + first.join(timeout=2) + second.join(timeout=2) + services.close() + + +def test_lsp_cancellation_invalidates_then_cleans_before_it_escapes(tmp_path): + project = make_lake_project(tmp_path, "lsp-cancellation") + source = project / "Main.lean" + source.write_text("#check Nat\n") + events = [] + services = None + + class Cancellation(BaseException): + pass + + class Session(FakeLsp): + def get_diagnostics(self, file_path): + raise Cancellation("cancel diagnostics") + + def abort(self): + resident = services.lsp_projects.stats()["resident"] + events.append(f"abort:valid={resident[0]['valid']}") + super().abort() + + services = LeanRuntimeServices( + runtime_config(), + repl_factory=FakePool, + lsp_factory=Session, + start_sweepers=False, + ) + try: + with pytest.raises(Cancellation, match="cancel diagnostics"): + services.dispatch( + "lsp.diagnostics", + {"project_dir": str(project), "file_path": str(source)}, + ) + assert events == ["abort:valid=False"] + finally: + services.close() + + +def test_lsp_operation_cancellation_remains_primary_after_cleanup_cancellation( + tmp_path, monkeypatch +): + from servers.lsp import server as lsp + + project = make_lake_project(tmp_path, "lsp-double-cancellation") + source = project / "Main.lean" + source.write_text("#check Nat\n") + + class OperationCancellation(BaseException): + pass + + class CleanupCancellation(BaseException): + pass + + operation_error = OperationCancellation("cancel operation") + cleanup_error = CleanupCancellation("cancel cleanup") + attempts = 0 + sessions = [] + + class Process: + pid = 51000 + returncode = None + stdin = None + stdout = None + stderr = None + + def poll(self): + return self.returncode + + def wait(self, timeout=None): + return self.returncode + + class Session(lsp.LeanLspSession): + def __init__(self, root): + super().__init__(lsp.LspConfig(cwd=str(root))) + self.process = Process() + self._process_group_id = self.process.pid + self._group_retired = False + sessions.append(self) + + def get_diagnostics(self, file_path): + raise operation_error + + def close(self): + self.abort() + + def retire(process, process_group_id): + nonlocal attempts + attempts += 1 + if attempts == 1: + raise cleanup_error + process.returncode = -9 + + monkeypatch.setattr(lsp, "_kill_subprocesses", retire) + services = LeanRuntimeServices( + runtime_config(), + repl_factory=FakePool, + lsp_factory=Session, + start_sweepers=False, + ) + try: + with pytest.raises(OperationCancellation, match="cancel operation") as raised: + services.dispatch( + "lsp.diagnostics", + {"project_dir": str(project), "file_path": str(source)}, + ) + assert raised.value is operation_error + assert raised.value.__cause__ is cleanup_error + assert attempts == 2 + assert sessions[0].is_clean() + finally: + services.close() + + +def test_runtime_close_cleans_every_lsp_before_reraising_cancellation(tmp_path): + first = make_lake_project(tmp_path, "first-lsp-close") + second = make_lake_project(tmp_path, "second-lsp-close") + cancellation = _RuntimeCloseCancellation("cancel first close") + sessions = [] + events = [] + + class Session(FakeLsp): + def __init__(self, root): + super().__init__(root) + self.number = len(sessions) + 1 + self.close_calls = 0 + sessions.append(self) + + def get_diagnostics(self, file_path): + return [] + + def close(self): + self.close_calls += 1 + self.closed = True + events.append(f"close:{self.number}") + if self.number == 1 and self.close_calls == 1: + raise cancellation + + services = LeanRuntimeServices( + runtime_config(), + repl_factory=FakePool, + lsp_factory=Session, + start_sweepers=False, + ) + for project in (first, second): + source = project / "Main.lean" + source.write_text("#check Nat\n") + services.dispatch( + "lsp.diagnostics", + {"project_dir": str(project), "file_path": str(source)}, + ) + + with pytest.raises(_RuntimeCloseCancellation, match="cancel first close"): + services.close() + + assert events == ["close:1", "close:1", "close:2"] + assert all(session.closed for session in sessions) + + +def test_runtime_attempts_lsp_cleanup_after_repl_cleanup_cancellation(tmp_path): + project = make_lake_project(tmp_path, "repl-close-cancellation") + source = project / "Main.lean" + source.write_text("#check Nat\n") + cancellation = _RuntimeCloseCancellation("cancel REPL close") + pools = [] + sessions = [] + + class Pool(FakePool): + def __init__(self, root): + super().__init__(root) + self.shutdown_calls = 0 + + def shutdown(self): + self.shutdown_calls += 1 + self._shutdown = True + if self.shutdown_calls == 1: + raise cancellation + + class Session(FakeLsp): + def get_diagnostics(self, file_path): + return [] + + services = LeanRuntimeServices( + runtime_config(), + repl_factory=lambda root: pools.append(Pool(root)) or pools[-1], + lsp_factory=lambda root: sessions.append(Session(root)) or sessions[-1], + start_sweepers=False, + ) + services.dispatch( + "repl.run", + {"project_dir": str(project), "code": "#check Nat", "timeout": None}, + ) + services.dispatch( + "lsp.diagnostics", + {"project_dir": str(project), "file_path": str(source)}, + ) + + with pytest.raises(_RuntimeCloseCancellation, match="cancel REPL close"): + services.close() + + assert pools[0]._shutdown is True + assert pools[0].shutdown_calls == 2 + assert sessions[0].closed is True + + +def test_stale_resource_close_cancellation_releases_creation_reservation(tmp_path): + project = make_lake_project(tmp_path, "stale-close-cancellation") + cancellation = _RuntimeCloseCancellation("cancel stale close") + resources = [] + + class Resource: + def __init__(self): + self.valid = True + self.close_calls = 0 + self.interrupt_close = not resources + resources.append(self) + + def close(self): + self.close_calls += 1 + self.valid = False + if self.interrupt_close and self.close_calls == 1: + raise cancellation + + cache = ProjectResourceCache( + lambda root: Resource(), + lambda resource: resource.close(), + max_entries=1, + idle_seconds=1800, + is_valid=lambda resource: resource.valid, + start_sweeper=False, + ) + with cache.lease(str(project)) as first: + first.valid = False + + with pytest.raises(_RuntimeCloseCancellation, match="cancel stale close"): + with cache.lease(str(project)): + pytest.fail("replacement must wait for the close cancellation to escape") + + assert cache.state(str(project)) == "cold" + assert resources[0].close_calls == 2 + with cache.lease(str(project)) as replacement: + assert replacement is resources[1] + cache.close() + + +class _RuntimeCloseCancellation(BaseException): + pass + + +@pytest.mark.skipif(not hasattr(os, "getuid"), reason="shared runtime requires POSIX locks") +def test_runtime_holds_its_lifetime_lock_through_lsp_cleanup(runtime_dir, monkeypatch): + import fcntl + + from servers import lean_runtime + from servers.lean_client import runtime_paths_for_socket + + paths = runtime_paths_for_socket(runtime_dir / "runtime.sock") + cleanup_started = threading.Event() + release_cleanup = threading.Event() + errors = [] + events = [] + + class Services: + def close(self): + events.append("cleanup:start") + cleanup_started.set() + assert release_cleanup.wait(timeout=2) + events.append("cleanup:end") + + class Server: + def __init__(self, socket_path, services): + self.socket_path = socket_path + self.services = services + socket_path.touch() + + def serve_forever(self, poll_interval): + events.append("serve") + + def server_close(self): + events.append("server:close") + + def request_shutdown(self): + pass + + monkeypatch.setattr(lean_runtime, "LeanRuntimeServices", Services) + monkeypatch.setattr(lean_runtime, "LeanRuntimeServer", Server) + monkeypatch.setattr(lean_runtime, "_configure_logging", lambda path: None) + monkeypatch.setattr(lean_runtime.signal, "signal", lambda signum, handler: None) + + def run() -> None: + try: + lean_runtime.serve(paths) + except BaseException as error: + errors.append(error) + + runtime = threading.Thread(target=run) + runtime.start() + contender = None + try: + assert cleanup_started.wait(timeout=1) + contender = os.open(paths.lifetime_lock, os.O_CREAT | os.O_RDWR, 0o600) + with pytest.raises(BlockingIOError): + fcntl.flock(contender, fcntl.LOCK_EX | fcntl.LOCK_NB) + assert runtime.is_alive() + + release_cleanup.set() + runtime.join(timeout=2) + assert not runtime.is_alive() + assert errors == [] + fcntl.flock(contender, fcntl.LOCK_EX | fcntl.LOCK_NB) + assert events == ["serve", "server:close", "cleanup:start", "cleanup:end"] + finally: + release_cleanup.set() + runtime.join(timeout=2) + if contender is not None: + os.close(contender) From b85255f02df650b018d07ee87838727106010f78 Mon Sep 17 00:00:00 2001 From: Jack McCarthy Date: Mon, 5 Oct 2026 07:26:29 -0400 Subject: [PATCH 02/10] Make LSP deadline test clock deterministic --- tests/test_lsp_protocol.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/tests/test_lsp_protocol.py b/tests/test_lsp_protocol.py index 39edf1b4b..47151d6fd 100644 --- a/tests/test_lsp_protocol.py +++ b/tests/test_lsp_protocol.py @@ -552,15 +552,20 @@ def test_document_close_budget_exhaustion_poisons_the_session(tmp_path, monkeypa source = tmp_path / "Test.lean" source.write_text("#check Nat\n") session = lsp.LeanLspSession(lsp.LspConfig(timeout=60)) - clock = iter([0.0, 0.0, 0.0, 0.0, 60.0]) + clock = {"now": 0.0} notifications = [] - monkeypatch.setattr(lsp.time, "monotonic", lambda: next(clock, 60.0)) + monkeypatch.setattr(lsp.time, "monotonic", lambda: clock["now"]) monkeypatch.setattr( session, "_send_notification", lambda method, params, **kwargs: notifications.append(method), ) - monkeypatch.setattr(session, "_collect_diagnostics", lambda uri, timeout: []) + + def finish_at_deadline(uri, timeout): + clock["now"] = 60.0 + return [] + + monkeypatch.setattr(session, "_collect_diagnostics", finish_at_deadline) with pytest.raises(TimeoutError, match="before closing the document"): session.get_diagnostics(str(source)) From 2fc66f479769f198a0fe789fed7a08e6d537140f Mon Sep 17 00:00:00 2001 From: Jack McCarthy Date: Mon, 5 Oct 2026 18:50:12 -0400 Subject: [PATCH 03/10] Golf LSP ownership down to group retirement and poisoning Keep the fix for #108 and drop machinery for states the daemon cannot reach: - Start the LSP in its own session and retire the whole process group through _kill_subprocesses. Cleanup retries until the group is verified gone instead of raising, so the runtime cannot release its lifetime lock or a cache slot while Lean workers survive. - Poison the session before a failed operation releases admission, so a queued call never reads the leftovers of a desynchronized stream. - A didClose failure or an exhausted budget poisons the session but keeps the result, preserving the existing response contract. - The runtime invalidates the cached session before aborting it on any non-busy failure. Removed as unreachable: async-cancellation bookkeeping (the daemon's signal handlers do not raise), LspCleanupError and its retry constants, and the ProjectResourceCache/services close changes, which the retry-until-clean abort makes redundant. --- servers/lean_runtime.py | 136 +------- servers/lsp/server.py | 462 ++++++------------------- tests/test_lsp_protocol.py | 549 +++++------------------------- tests/test_shared_lean_runtime.py | 304 ++--------------- 4 files changed, 218 insertions(+), 1233 deletions(-) diff --git a/servers/lean_runtime.py b/servers/lean_runtime.py index 39d702b72..c20552815 100644 --- a/servers/lean_runtime.py +++ b/servers/lean_runtime.py @@ -421,11 +421,9 @@ def close(self) -> None: resources = [entry.resource for entry in self._entries.values()] self._entries.clear() self._condition.notify_all() - try: - self._close_many(resources) - finally: - if self._sweeper and self._sweeper is not threading.current_thread(): - self._sweeper.join() + self._close_many(resources) + if self._sweeper and self._sweeper is not threading.current_thread(): + self._sweeper.join() def _acquire( self, @@ -544,29 +542,12 @@ def _acquire( self._condition.wait(timeout=wait_seconds) if resources_to_close: - try: - self._close_many(resources_to_close) - except BaseException: - with self._condition: - if reserved: - self._creating.discard(root) - self._condition.notify_all() - raise - finally: - resources_to_close.clear() - - if resources_to_close: - try: self._close_many(resources_to_close) - except BaseException: - with self._condition: - if reserved: - self._creating.discard(root) - self._condition.notify_all() - raise - finally: resources_to_close.clear() + if resources_to_close: + self._close_many(resources_to_close) + if not reserved: return resource @@ -635,22 +616,8 @@ def _sweep(self, interval: float) -> None: logger.exception("failed to evict idle Lean project resources") def _close_many(self, resources: list[T]) -> None: - first_error: BaseException | None = None - index = 0 - while index < len(resources): - try: - while index < len(resources): - self._safe_close(resources[index]) - index += 1 - except BaseException as error: - if first_error is None: - first_error = error - elif error is not first_error: - add_note = getattr(first_error, "add_note", None) - if add_note is not None: - add_note(f"another Lean resource close was interrupted by: {error}") - if first_error is not None: - raise first_error.with_traceback(first_error.__traceback__) + for resource in resources: + self._safe_close(resource) def _safe_close(self, resource: T) -> None: try: @@ -695,25 +662,7 @@ def default_lsp_factory(project_dir: Path) -> LeanLspSession: timeout=self.config.lsp_timeout, ) ) - try: - session.start() - except BaseException as start_error: - if not session.is_clean(): - try: - session.abort() - except BaseException as cleanup_error: - if not isinstance(start_error, Exception): - add_note = getattr(start_error, "add_note", None) - if add_note is not None: - add_note( - "Lean LSP startup cleanup was also interrupted by: " - f"{cleanup_error}" - ) - raise start_error.with_traceback( - start_error.__traceback__ - ) from cleanup_error - raise cleanup_error from start_error - raise + session.start() return session self.repl_projects = ProjectResourceCache( @@ -796,24 +745,9 @@ def dispatch(self, method: str, params: dict[str, Any]) -> Any: diagnostics = session.get_diagnostics(str(path)) except LspBusyError: raise - except BaseException as operation_error: + except Exception: self.lsp_projects.invalidate(str(root), session) - try: - session.abort() - except BaseException as cleanup_error: - if cleanup_error is operation_error: - raise - if not isinstance(operation_error, Exception): - add_note = getattr(operation_error, "add_note", None) - if add_note is not None: - add_note( - "Lean LSP cleanup was also interrupted by: " - f"{cleanup_error}" - ) - raise operation_error.with_traceback( - operation_error.__traceback__ - ) from cleanup_error - raise cleanup_error from operation_error + session.abort() raise return format_lsp_diagnostics(diagnostics) if method == "lsp.hover": @@ -834,24 +768,9 @@ def dispatch(self, method: str, params: dict[str, Any]) -> Any: result = session.hover(str(path), line, character) except LspBusyError: raise - except BaseException as operation_error: + except Exception: self.lsp_projects.invalidate(str(root), session) - try: - session.abort() - except BaseException as cleanup_error: - if cleanup_error is operation_error: - raise - if not isinstance(operation_error, Exception): - add_note = getattr(operation_error, "add_note", None) - if add_note is not None: - add_note( - "Lean LSP cleanup was also interrupted by: " - f"{cleanup_error}" - ) - raise operation_error.with_traceback( - operation_error.__traceback__ - ) from cleanup_error - raise cleanup_error from operation_error + session.abort() raise return result or "No hover information at this position." raise ValueError(f"unknown Lean runtime method: {method}") @@ -872,33 +791,8 @@ def status(self, *, include_projects: bool) -> dict[str, Any]: return result def close(self) -> None: - close_error: BaseException | None = None - lsp_close_error: BaseException | None = None - try: - self.repl_projects.close() - except BaseException as error: - close_error = error - finally: - while True: - try: - self.lsp_projects.close() - except BaseException as error: - if lsp_close_error is None: - lsp_close_error = error - elif error is not lsp_close_error: - add_note = getattr(lsp_close_error, "add_note", None) - if add_note is not None: - add_note(f"Lean LSP cleanup was also interrupted by: {error}") - continue - break - if close_error is None: - close_error = lsp_close_error - elif lsp_close_error is not None and lsp_close_error is not close_error: - add_note = getattr(close_error, "add_note", None) - if add_note is not None: - add_note(f"Lean LSP cleanup was also interrupted by: {lsp_close_error}") - if close_error is not None: - raise close_error.with_traceback(close_error.__traceback__) + self.repl_projects.close() + self.lsp_projects.close() def _acquisition_timeout(self, operation_timeout: float) -> float: """Reserve enough of the RPC deadline for the admitted tool operation.""" diff --git a/servers/lsp/server.py b/servers/lsp/server.py index c5bac6995..0c9c4d9c9 100644 --- a/servers/lsp/server.py +++ b/servers/lsp/server.py @@ -28,8 +28,6 @@ DEFAULT_LSP_TIMEOUT = 60 MAX_LSP_HEADER_BYTES = 16 * 1024 MAX_LSP_MESSAGE_BYTES = 16 * 1024 * 1024 -LSP_CLEANUP_RETRY_INITIAL_SECONDS = 0.01 -LSP_CLEANUP_RETRY_MAX_SECONDS = 1.0 class LspProtocolError(RuntimeError): @@ -40,10 +38,6 @@ class LspBusyError(TimeoutError): """A queued operation could not enter the shared LSP session in time.""" -class LspCleanupError(LspProtocolError): - """The owned language-server process group could not be retired.""" - - @dataclass class LspConfig: """Configuration for the Lean LSP server.""" @@ -60,10 +54,8 @@ def __init__(self, config: LspConfig) -> None: self.config = config self.process: subprocess.Popen | None = None self._process_group_id: int | None = None - self._group_retired = True + # Set once the protocol stream may be desynchronized; never cleared. self._poisoned = False - self._retire_pending = False - self._pending_cancellation: BaseException | None = None self._request_id = 0 self._lock = threading.Lock() self._lifecycle_lock = threading.Lock() @@ -76,312 +68,89 @@ def start(self) -> None: """Start the language server process.""" if os.name != "posix": raise RuntimeError("Lean LSP transport requires a POSIX platform") - spawned_process: subprocess.Popen | None = None - try: - with self._lifecycle_lock: - if not self.is_clean(): - raise RuntimeError("Lean LSP session already owns a process") - env = os.environ.copy() - env.pop("PYTHONPATH", None) - try: - spawned_process = subprocess.Popen( - self.config.lake_command, - cwd=self.config.cwd, - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - env=env, - start_new_session=True, - ) - self.process = spawned_process - self._group_retired = False - self._poisoned = True - self._retire_pending = True - self._publish_process_group(spawned_process) - - # An initialize response must be an InitializeResult object. A - # timeout, JSON-RPC error, or malformed result means the backing - # Lean server is unusable, so do not expose an apparently healthy - # MCP server on top of it. - result = self._send_request("initialize", { - "processId": os.getpid(), - "capabilities": {}, - "rootUri": Path(self.config.cwd).resolve().as_uri(), - }) - if not isinstance(result, dict): - raise LspProtocolError( - "LSP initialize returned a non-object result: " - f"{result!r}" - ) + env = os.environ.copy() + env.pop("PYTHONPATH", None) + + self.process = subprocess.Popen( + self.config.lake_command, + cwd=self.config.cwd, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + env=env, + start_new_session=True, + ) + self._process_group_id = self.process.pid - self._send_notification("initialized", {}) - except BaseException as start_error: - if self.process is None and spawned_process is not None: - self.process = spawned_process - if spawned_process is not None and self._process_group_id is None: - self._group_retired = False - self._poisoned = True - self._retire_pending = True - self._remember_cancellation(start_error) - try: - self._retire_until_clean("after startup failure") - except BaseException as cleanup_error: - if cleanup_error is start_error: - raise - raise cleanup_error from start_error - raise - except BaseException as start_error: - if self.is_clean(): - raise - if self.process is None and spawned_process is not None: - self.process = spawned_process - self._group_retired = False - self._poisoned = True - self._retire_pending = True - self._recover_interrupted_retirement( - start_error, - "after interrupted startup", - ) + try: + # An initialize response must be an InitializeResult object. A + # timeout, JSON-RPC error, or malformed result means the backing + # Lean server is unusable, so do not expose an apparently healthy + # MCP server on top of it. + result = self._send_request("initialize", { + "processId": os.getpid(), + "capabilities": {}, + "rootUri": Path(self.config.cwd).resolve().as_uri(), + }) + if not isinstance(result, dict): + raise LspProtocolError( + "LSP initialize returned a non-object result: " + f"{result!r}" + ) - def _publish_process_group(self, process: subprocess.Popen) -> None: - """Publish complete ownership immediately after a successful spawn.""" - self._process_group_id = process.pid - self._group_retired = False - self._poisoned = False - self._retire_pending = False + self._send_notification("initialized", {}) + except BaseException: + self._abort_process() + raise def _abort_process(self) -> None: - """Force-close the complete owned process group and verify cleanup.""" - if self.is_clean(): - self._raise_pending_cancellation() - return + """Retire the whole process group, retaining ownership until verified. + + Retries instead of raising: once this returns, callers release the + session and the runtime releases its lifetime lock. + """ self._poisoned = True - self._retire_pending = True - self._complete_retirement("after abort") + delay = 0.01 + while self.process is not None: + try: + if self._process_group_id is not None: + _kill_subprocesses(self.process, self._process_group_id) + # The group id may be reused once the leader is reaped. + self._process_group_id = None + for stream in (self.process.stdin, self.process.stdout): + if stream is not None: + stream.close() + self.process = None + except Exception: + logger.exception("failed to retire the Lean LSP process group; retrying") + time.sleep(delay) + delay = min(delay * 2, 1.0) def close(self) -> None: """Shut down the language server.""" - try: - with self._lifecycle_lock: + with self._lifecycle_lock: + if self.is_alive(): try: - self._poisoned = True - if self.is_clean(): - self._raise_pending_cancellation() - return - process = self.process - already_retiring = self._retire_pending - self._retire_pending = True - if not already_retiring and process is not None: - if process.poll() is None: - self._send_request("shutdown", {}) - self._send_notification("exit", {}) - process.wait(timeout=5) - except BaseException as error: - self._remember_cancellation(error) - self._poisoned = True - if not self.is_clean(): - self._retire_pending = True - if isinstance(error, Exception): - logger.warning( - "graceful Lean LSP shutdown failed; enforcing cleanup", - exc_info=True, - ) - self._complete_retirement("during close") - except BaseException as error: - if self.is_clean(): - raise - self._recover_interrupted_retirement(error, "during interrupted close") + self._send_request("shutdown", {}) + self._send_notification("exit", {}) + self.process.wait(timeout=5) + except Exception: + logger.warning("graceful Lean LSP shutdown failed", exc_info=True) + # A graceful wrapper exit can still leave Lean workers in the group. + self._abort_process() def abort(self) -> None: """Discard a protocol stream that can no longer be shared safely.""" - try: - with self._lifecycle_lock: - try: - self._poisoned = True - self._abort_process() - except BaseException as error: - if self.is_clean(): - raise - self._remember_cancellation(error) - self._poisoned = True - self._retire_pending = True - self._complete_retirement("after interrupted abort") - except BaseException as error: - if self.is_clean(): - raise - self._recover_interrupted_retirement(error, "after interrupted abort") + with self._lifecycle_lock: + self._abort_process() def is_alive(self) -> bool: """Return whether the cached language-server child can accept work.""" return ( self.process is not None and not self._poisoned - and not self._retire_pending and self.process.poll() is None ) - def is_clean(self) -> bool: - """Return whether this object owns no process, group, or open pipes.""" - return ( - self.process is None - and self._process_group_id is None - and not self._retire_pending - ) - - def _mark_retire_pending(self) -> None: - """Poison the protocol stream before another admitted operation sees it.""" - self._poisoned = True - self._retire_pending = True - - def _require_usable_after_admission(self) -> None: - if self._poisoned or self._retire_pending: - raise LspProtocolError("Lean LSP session is retiring after a failed operation") - - def _remember_cancellation(self, error: BaseException) -> None: - if isinstance(error, Exception): - return - if self._pending_cancellation is None: - self._pending_cancellation = error - return - add_note = getattr(self._pending_cancellation, "add_note", None) - if add_note is not None and error is not self._pending_cancellation: - add_note(f"Lean LSP cleanup was also interrupted by: {error}") - - def _raise_pending_cancellation(self) -> None: - error, self._pending_cancellation = self._pending_cancellation, None - if error is not None: - raise error.with_traceback(error.__traceback__) - - @staticmethod - def _close_process_streams(process: subprocess.Popen) -> None: - first_error: Exception | None = None - first_cancellation: BaseException | None = None - for name in ("stdin", "stdout", "stderr"): - stream = getattr(process, name, None) - if stream is None: - continue - try: - stream.close() - except BaseException as error: - if isinstance(error, Exception): - if first_error is None: - first_error = error - elif first_cancellation is None: - first_cancellation = error - else: - add_note = getattr(first_cancellation, "add_note", None) - if add_note is not None: - add_note(f"Lean LSP pipe cleanup was also interrupted by: {error}") - else: - setattr(process, name, None) - if first_cancellation is not None: - if first_error is not None: - add_note = getattr(first_cancellation, "add_note", None) - if add_note is not None: - add_note(f"Lean LSP pipe cleanup also failed: {first_error}") - raise first_cancellation - if first_error is not None: - raise LspCleanupError("failed to close Lean LSP process pipes") from first_error - - def _retire_owned_process_once(self) -> None: - process = self.process - process_group_id = self._process_group_id - if process is None and process_group_id is None: - self._retire_pending = False - return - if process is None: - raise LspCleanupError("Lean LSP process ownership is incomplete") - if not self._group_retired: - if process_group_id is None: - try: - process_group_id = process.pid - except Exception as error: - raise LspCleanupError( - "Lean LSP process-group id is unavailable" - ) from error - self._process_group_id = process_group_id - try: - assert process_group_id is not None - _kill_subprocesses(process, process_group_id) - except Exception as error: - raise LspCleanupError( - "failed to retire the Lean LSP process group" - ) from error - # Publish verified retirement before clearing the PGID. If a - # cancellation lands between these stores, the retry sees this - # flag and cannot signal a subsequently reused process-group id. - self._group_retired = True - self._process_group_id = None - self._close_process_streams(process) - if all(getattr(process, name, None) is None for name in ("stdin", "stdout", "stderr")): - self.process = None - self._retire_pending = False - - def _retire_owned_process(self) -> None: - """Retry cancellation-like interruptions, but expose cleanup failures.""" - while not self.is_clean(): - try: - self._retire_owned_process_once() - except Exception: - raise - except BaseException as error: - self._remember_cancellation(error) - self._raise_pending_cancellation() - - def _retire_until_clean(self, context: str) -> None: - """Retain ownership and retry until every child and pipe is clean.""" - delay = LSP_CLEANUP_RETRY_INITIAL_SECONDS - while not self.is_clean(): - try: - self._retire_owned_process() - except LspCleanupError: - logger.exception("Lean LSP cleanup failed %s; retrying", context) - except BaseException: - # `_retire_owned_process` releases a remembered cancellation only - # after cleanup is verified. - if self.is_clean(): - raise - raise - if self.is_clean(): - break - try: - time.sleep(delay) - except BaseException as error: - self._remember_cancellation(error) - delay = min(delay * 2, LSP_CLEANUP_RETRY_MAX_SECONDS) - self._raise_pending_cancellation() - - def _complete_retirement(self, context: str) -> None: - """Do not let an interruption escape while this object still owns state.""" - while not self.is_clean(): - try: - self._retire_until_clean(context) - except BaseException as error: - if self.is_clean(): - raise - self._remember_cancellation(error) - if isinstance(error, Exception): - logger.exception("Lean LSP retirement failed %s; retrying", context) - self._raise_pending_cancellation() - - def _recover_interrupted_retirement( - self, - error: BaseException, - context: str, - ) -> None: - """Reacquire lifecycle ownership until cancellation-safe cleanup finishes.""" - self._remember_cancellation(error) - while not self.is_clean(): - try: - with self._lifecycle_lock: - self._poisoned = True - self._retire_pending = True - self._complete_retirement(context) - except BaseException as retry_error: - if self.is_clean(): - raise - self._remember_cancellation(retry_error) - self._raise_pending_cancellation() - def get_diagnostics(self, file_path: str) -> list[dict]: """Open a file and collect diagnostics from the language server.""" deadline = time.monotonic() + self.config.timeout @@ -390,7 +159,8 @@ def get_diagnostics(self, file_path: str) -> list[dict]: f"timed out after {self.config.timeout:g}s waiting for the Lean LSP session" ) try: - self._require_usable_after_admission() + if self._poisoned: + raise LspProtocolError("Lean LSP session is retiring after a failed operation") remaining = deadline - time.monotonic() if remaining <= 0: raise LspBusyError( @@ -399,7 +169,9 @@ def get_diagnostics(self, file_path: str) -> list[dict]: try: return self._get_diagnostics(file_path, timeout=remaining) except BaseException: - self._mark_retire_pending() + # Poison before releasing admission so a queued call cannot + # read the failed operation's leftovers from the stream. + self._poisoned = True raise finally: self._operation_lock.release() @@ -428,25 +200,30 @@ def _get_diagnostics(self, file_path: str, *, timeout: float | None = None) -> l timeout=self._remaining(deadline, operation_timeout), ) - diagnostics: list[dict] | None = None - operation_error: BaseException | None = None try: # An empty published diagnostic list means the file is clean. No # publication at all is a timeout/error and must never be conflated # with that valid empty result. - diagnostics = self._collect_diagnostics( + return self._collect_diagnostics( uri, timeout=self._remaining(deadline, operation_timeout), ) - except BaseException as error: - operation_error = error - close_error = self._close_document(uri, deadline) - if operation_error is not None: - self._raise_operation_error(operation_error, close_error) - if close_error is not None: - raise close_error - assert diagnostics is not None - return diagnostics + finally: + try: + remaining = deadline - time.monotonic() + if remaining > 0: + self._send_notification( + "textDocument/didClose", + {"textDocument": {"uri": uri}}, + timeout=remaining, + ) + else: + raise TimeoutError("no time remained to close the document") + except Exception: + # An unclosed document or a partial frame desynchronizes the + # stream; keep this result but never reuse the session. + self._poisoned = True + logger.warning("failed to close LSP document %s", uri, exc_info=True) def hover(self, file_path: str, line: int, character: int) -> str | None: """Get hover information at a position.""" @@ -456,7 +233,8 @@ def hover(self, file_path: str, line: int, character: int) -> str | None: f"timed out after {self.config.timeout:g}s waiting for the Lean LSP session" ) try: - self._require_usable_after_admission() + if self._poisoned: + raise LspProtocolError("Lean LSP session is retiring after a failed operation") remaining = deadline - time.monotonic() if remaining <= 0: raise LspBusyError( @@ -465,7 +243,9 @@ def hover(self, file_path: str, line: int, character: int) -> str | None: try: return self._hover(file_path, line, character, timeout=remaining) except BaseException: - self._mark_retire_pending() + # Poison before releasing admission so a queued call cannot + # read the failed operation's leftovers from the stream. + self._poisoned = True raise finally: self._operation_lock.release() @@ -494,8 +274,6 @@ def _hover( }, timeout=self._remaining(deadline, timeout), ) - result: Any = None - operation_error: BaseException | None = None try: result = self._send_request( "textDocument/hover", @@ -505,13 +283,22 @@ def _hover( }, timeout=self._remaining(deadline, timeout), ) - except BaseException as error: - operation_error = error - close_error = self._close_document(uri, deadline) - if operation_error is not None: - self._raise_operation_error(operation_error, close_error) - if close_error is not None: - raise close_error + finally: + try: + remaining = deadline - time.monotonic() + if remaining > 0: + self._send_notification( + "textDocument/didClose", + {"textDocument": {"uri": uri}}, + timeout=remaining, + ) + else: + raise TimeoutError("no time remained to close the document") + except Exception: + # An unclosed document or a partial frame desynchronizes the + # stream; keep this result but never reuse the session. + self._poisoned = True + logger.warning("failed to close LSP document %s", uri, exc_info=True) if result and "contents" in result: contents = result["contents"] if isinstance(contents, dict): @@ -519,43 +306,6 @@ def _hover( return str(contents) return None - def _close_document(self, uri: str, deadline: float) -> BaseException | None: - remaining = deadline - time.monotonic() - if remaining <= 0: - return TimeoutError( - "Lean LSP operation deadline expired before closing the document" - ) - try: - self._send_notification( - "textDocument/didClose", - {"textDocument": {"uri": uri}}, - timeout=remaining, - ) - except BaseException as error: - return error - return None - - @staticmethod - def _raise_operation_error( - operation_error: BaseException, - close_error: BaseException | None, - ) -> None: - if close_error is not None and close_error is not operation_error: - if isinstance(operation_error, Exception) and not isinstance( - close_error, Exception - ): - add_note = getattr(close_error, "add_note", None) - if add_note is not None: - add_note(f"the Lean LSP operation also failed: {operation_error}") - raise close_error.with_traceback(close_error.__traceback__) from operation_error - add_note = getattr(operation_error, "add_note", None) - if add_note is not None: - add_note(f"closing the Lean LSP document also failed: {close_error}") - raise operation_error.with_traceback( - operation_error.__traceback__ - ) from close_error - raise operation_error.with_traceback(operation_error.__traceback__) - def _send_request(self, method: str, params: dict, *, timeout: float = 30) -> Any: """Send a JSON-RPC request and wait for response.""" deadline = time.monotonic() + timeout diff --git a/tests/test_lsp_protocol.py b/tests/test_lsp_protocol.py index 47151d6fd..89ddee8ac 100644 --- a/tests/test_lsp_protocol.py +++ b/tests/test_lsp_protocol.py @@ -15,15 +15,26 @@ from servers.lsp import server as lsp +class _FakeStream: + def __init__(self, *, failures: int = 0) -> None: + self.failures = failures + self.close_calls = 0 + + def close(self) -> None: + self.close_calls += 1 + if self.failures: + self.failures -= 1 + raise OSError("cannot close pipe") + + class _FakeProcess: - def __init__(self, *, pid: int = 43210) -> None: - self.pid = pid + def __init__(self) -> None: + self.pid = 43210 self.returncode: int | None = None self.killed = False self.waited = False self.stdin = _FakeStream() self.stdout = _FakeStream() - self.stderr = None def poll(self): return self.returncode @@ -37,40 +48,17 @@ def wait(self, timeout=None): return self.returncode -class _FakeStream: - def __init__( - self, - *, - failures: int = 0, - errors: list[BaseException] | None = None, - events: list[str] | None = None, - name: str = "", - ) -> None: - self.failures = failures - self.errors = list(errors or []) - self.events = events - self.name = name - self.close_calls = 0 - - def close(self) -> None: - self.close_calls += 1 - if self.events is not None: - self.events.append(f"close:{self.name}") - if self.failures: - self.failures -= 1 - raise OSError(f"cannot close {self.name}") - if self.errors: - raise self.errors.pop(0) - +def _retire_fake_group(process: _FakeProcess, process_group_id: int) -> None: + assert process_group_id == process.pid + process.kill() + process.wait() -class _Cancellation(BaseException): - pass - -def _finish_fake_process(process: _FakeProcess) -> None: - process.killed = True - process.returncode = -9 - process.waited = True +def _owned_session(process: _FakeProcess) -> lsp.LeanLspSession: + session = lsp.LeanLspSession(lsp.LspConfig()) + session.process = process + session._process_group_id = process.pid + return session def test_json_rpc_error_response_raises_protocol_error(monkeypatch): @@ -98,11 +86,7 @@ def popen(*args, **kwargs): return process monkeypatch.setattr(lsp.subprocess, "Popen", popen) - monkeypatch.setattr( - lsp, - "_kill_subprocesses", - lambda owned, process_group_id: _finish_fake_process(owned), - ) + monkeypatch.setattr(lsp, "_kill_subprocesses", _retire_fake_group) session = lsp.LeanLspSession(lsp.LspConfig()) @@ -114,12 +98,12 @@ def fail_initialize(method, params): with pytest.raises(lsp.LspProtocolError, match="initialize rejected"): session.start() + assert popen_kwargs["start_new_session"] is True assert process.killed is True assert process.waited is True - assert popen_kwargs["start_new_session"] is True + assert process.stdin.close_calls == process.stdout.close_calls == 1 assert session.process is None assert session._process_group_id is None - assert session.is_clean() def test_start_rejects_an_unsupported_platform_before_spawning(monkeypatch): @@ -145,311 +129,57 @@ def retire(owned, process_group_id): attempts += 1 if attempts == 1: raise RuntimeError("stubborn process group") - _finish_fake_process(owned) - - monkeypatch.setattr(lsp, "_kill_subprocesses", retire) - session = lsp.LeanLspSession(lsp.LspConfig()) - monkeypatch.setattr( - session, - "_send_request", - lambda method, params: (_ for _ in ()).throw( - lsp.LspProtocolError("initialize rejected") - ), - ) - - with pytest.raises(lsp.LspProtocolError, match="initialize rejected"): - session.start() - - assert attempts == 2 - assert session.is_clean() - - -def test_startup_publication_cancellation_still_retires_the_uncached_child(monkeypatch): - process = _FakeProcess() - session = lsp.LeanLspSession(lsp.LspConfig()) - monkeypatch.setattr(lsp.subprocess, "Popen", lambda *args, **kwargs: process) - monkeypatch.setattr( - session, - "_publish_process_group", - lambda owned: (_ for _ in ()).throw( - _Cancellation("cancel process-group publication") - ), - ) - retired = [] - - def retire(owned, process_group_id): - retired.append((owned, process_group_id)) - _finish_fake_process(owned) + _retire_fake_group(owned, process_group_id) monkeypatch.setattr(lsp, "_kill_subprocesses", retire) - - with pytest.raises(_Cancellation, match="cancel process-group publication"): - session.start() - - assert retired == [(process, process.pid)] - assert session.is_clean() - - -def test_startup_lock_exit_cancellation_retires_the_initialized_child(monkeypatch): - process = _FakeProcess() session = lsp.LeanLspSession(lsp.LspConfig()) - inner_lock = threading.Lock() - - class InterruptingLock: - def __init__(self): - self.interrupt = True - - def __enter__(self): - inner_lock.acquire() - return self - - def __exit__(self, exc_type, exc_value, traceback): - inner_lock.release() - if self.interrupt: - self.interrupt = False - raise _Cancellation("cancel lifecycle-lock exit") - return False - - session._lifecycle_lock = InterruptingLock() - monkeypatch.setattr(lsp.subprocess, "Popen", lambda *args, **kwargs: process) - monkeypatch.setattr(session, "_send_request", lambda method, params: {}) - monkeypatch.setattr(session, "_send_notification", lambda method, params: None) - retired = [] - def retire(owned, process_group_id): - retired.append((owned, process_group_id)) - _finish_fake_process(owned) + def fail_initialize(method, params): + raise lsp.LspProtocolError("initialize rejected") - monkeypatch.setattr(lsp, "_kill_subprocesses", retire) + monkeypatch.setattr(session, "_send_request", fail_initialize) - with pytest.raises(_Cancellation, match="cancel lifecycle-lock exit"): + with pytest.raises(lsp.LspProtocolError, match="initialize rejected"): session.start() - assert retired == [(process, process.pid)] - assert session.is_clean() - - -def test_close_preserves_cancellation_until_process_and_pipes_are_clean(monkeypatch): - process = _FakeProcess() - stdin = process.stdin - stdout = process.stdout - session = lsp.LeanLspSession(lsp.LspConfig()) - session.process = process - session._process_group_id = process.pid - session._group_retired = False - events = [] - - def cancel_shutdown(method, params): - events.append("cancel") - raise _Cancellation("cancel shutdown") - - def retire(owned, process_group_id): - events.append("retire") - _finish_fake_process(owned) - - monkeypatch.setattr(session, "_send_request", cancel_shutdown) - monkeypatch.setattr(lsp, "_kill_subprocesses", retire) - - with pytest.raises(_Cancellation, match="cancel shutdown"): - session.close() - - assert events == ["cancel", "retire"] - assert stdin.close_calls == 1 - assert stdout.close_calls == 1 - assert session.is_clean() + assert attempts == 2 + assert session.process is None -def test_cleanup_retry_retains_ownership_and_never_resignals_after_group_exit(monkeypatch): - events: list[str] = [] +def test_cleanup_retry_never_resignals_a_retired_group(monkeypatch): process = _FakeProcess() - process.stdin = _FakeStream(failures=1, events=events, name="stdin") - process.stdout = _FakeStream(events=events, name="stdout") - session = lsp.LeanLspSession(lsp.LspConfig()) - session.process = process - session._process_group_id = process.pid - session._group_retired = False - session._retire_pending = True - retire_calls = 0 + process.stdin = _FakeStream(failures=1) + session = _owned_session(process) + signals = [] def retire(owned, process_group_id): - nonlocal retire_calls - retire_calls += 1 - events.append("retire") - _finish_fake_process(owned) + signals.append(process_group_id) + _retire_fake_group(owned, process_group_id) monkeypatch.setattr(lsp, "_kill_subprocesses", retire) - monkeypatch.setattr(lsp.time, "sleep", lambda delay: events.append("retry")) + monkeypatch.setattr(lsp.time, "sleep", lambda delay: None) session.abort() - assert events == ["retire", "close:stdin", "close:stdout", "retry", "close:stdin"] - assert retire_calls == 1 - assert process.stdin is None - assert process.stdout is None - assert session.is_clean() - - -def test_cancellation_between_group_retirement_stores_never_resignals(monkeypatch): - class Session(lsp.LeanLspSession): - def __setattr__(self, name, value): - if ( - name == "_group_retired" - and value is True - and getattr(self, "interrupt_retirement_store", False) - ): - object.__setattr__(self, name, value) - object.__setattr__(self, "interrupt_retirement_store", False) - raise _Cancellation("cancel retirement publication") - super().__setattr__(name, value) - - process = _FakeProcess() - session = Session(lsp.LspConfig()) - session.process = process - session._process_group_id = process.pid - session._group_retired = False - session._retire_pending = True - session.interrupt_retirement_store = True - retire_calls = 0 - - def retire(owned, process_group_id): - nonlocal retire_calls - retire_calls += 1 - _finish_fake_process(owned) - - monkeypatch.setattr(lsp, "_kill_subprocesses", retire) - - with pytest.raises(_Cancellation, match="cancel retirement publication"): - session.abort() - - assert retire_calls == 1 - assert session.is_clean() - - -def test_pipe_cleanup_preserves_cancellation_over_an_ordinary_close_error(monkeypatch): - events: list[str] = [] - cancellation = _Cancellation("cancel pipe cleanup") - process = _FakeProcess() - process.stdin = _FakeStream(failures=1, events=events, name="stdin") - process.stdout = _FakeStream( - errors=[cancellation], - events=events, - name="stdout", - ) - stdin = process.stdin - stdout = process.stdout - session = lsp.LeanLspSession(lsp.LspConfig()) - session.process = process - session._process_group_id = process.pid - session._group_retired = False - session._retire_pending = True - retire_calls = 0 - - def retire(owned, process_group_id): - nonlocal retire_calls - retire_calls += 1 - events.append("retire") - _finish_fake_process(owned) - - monkeypatch.setattr(lsp, "_kill_subprocesses", retire) - - with pytest.raises(_Cancellation, match="cancel pipe cleanup"): - session.abort() - - assert retire_calls == 1 - assert session.is_clean() - assert process.stdin is None - assert process.stdout is None - assert stdin.close_calls == 2 - assert stdout.close_calls == 2 - - -def test_cleanup_retry_preserves_sleep_cancellation_until_cleanup_succeeds(monkeypatch): - process = _FakeProcess() - session = lsp.LeanLspSession(lsp.LspConfig()) - session.process = process - session._process_group_id = process.pid - session._group_retired = False - session._retire_pending = True - attempts = 0 - sleeps = 0 - - def retire(owned, process_group_id): - nonlocal attempts - attempts += 1 - if attempts == 1: - raise RuntimeError("retry cleanup") - _finish_fake_process(owned) - - def interrupt_sleep(delay): - nonlocal sleeps - sleeps += 1 - raise _Cancellation("cancel retry sleep") - - monkeypatch.setattr(lsp, "_kill_subprocesses", retire) - monkeypatch.setattr(lsp.time, "sleep", interrupt_sleep) - - with pytest.raises(_Cancellation, match="cancel retry sleep"): - session.abort() - - assert attempts == 2 - assert sleeps == 1 - assert session.is_clean() - - -@pytest.mark.parametrize("operation", ["abort", "close"]) -def test_retirement_transition_cannot_be_cancelled_before_cleanup( - monkeypatch, operation -): - process = _FakeProcess() - session = lsp.LeanLspSession(lsp.LspConfig()) - session.process = process - session._process_group_id = process.pid - session._group_retired = False - session._retire_pending = True - original_retire = session._retire_until_clean - calls = 0 - - def interrupt_before_cleanup(context): - nonlocal calls - calls += 1 - if calls == 1: - raise _Cancellation("cancel before cleanup") - return original_retire(context) - - monkeypatch.setattr(session, "_retire_until_clean", interrupt_before_cleanup) - monkeypatch.setattr( - lsp, - "_kill_subprocesses", - lambda owned, process_group_id: _finish_fake_process(owned), - ) - - with pytest.raises(_Cancellation, match="cancel before cleanup"): - getattr(session, operation)() - - assert calls == 2 - assert session.is_clean() + # The pipe failure is retried after the leader was reaped, when its + # process-group id may already belong to an unrelated group. + assert signals == [process.pid] + assert process.stdin.close_calls == 2 + assert session.process is None def test_failed_operation_poisons_a_queued_waiter_before_releasing_admission(monkeypatch): session = lsp.LeanLspSession(lsp.LspConfig()) first_entered = threading.Event() release_first = threading.Event() - second_touched_transport = threading.Event() + calls = [] errors: list[BaseException] = [] - calls = 0 - calls_lock = threading.Lock() def diagnostics(file_path, *, timeout=None): - nonlocal calls - with calls_lock: - calls += 1 - number = calls - if number == 1: - first_entered.set() - assert release_first.wait(timeout=2) - raise lsp.LspProtocolError("broken shared stream") - second_touched_transport.set() - return [] + calls.append(file_path) + first_entered.set() + assert release_first.wait(timeout=2) + raise lsp.LspProtocolError("broken shared stream") monkeypatch.setattr(session, "_get_diagnostics", diagnostics) @@ -468,84 +198,44 @@ def run() -> None: first.join(timeout=2) second.join(timeout=2) - assert not first.is_alive() - assert not second.is_alive() - assert not second_touched_transport.is_set() - assert len(errors) == 2 - messages = {str(error) for error in errors} - assert any("broken shared stream" in message for message in messages) - assert any("session is retiring" in message for message in messages) - - -def test_verified_cleanup_does_not_make_a_poisoned_session_reusable(monkeypatch): - session = lsp.LeanLspSession(lsp.LspConfig()) - session._mark_retire_pending() - session.abort() - assert session.is_clean() - assert not session.is_alive() - monkeypatch.setattr( - session, - "_get_diagnostics", - lambda *args, **kwargs: pytest.fail("a cleaned poisoned stream was reused"), - ) - - with pytest.raises(lsp.LspProtocolError, match="session is retiring"): - session.get_diagnostics("ignored.lean") + assert not first.is_alive() and not second.is_alive() + assert len(calls) == 1 + assert sorted(str(error) for error in errors) == [ + "Lean LSP session is retiring after a failed operation", + "broken shared stream", + ] @pytest.mark.parametrize("operation", ["diagnostics", "hover"]) -def test_did_close_failure_poisons_a_queued_operation( +def test_did_close_failure_keeps_the_result_but_poisons_the_session( tmp_path, monkeypatch, operation ): source = tmp_path / "Test.lean" source.write_text("#check Nat\n") session = lsp.LeanLspSession(lsp.LspConfig()) - close_entered = threading.Event() - release_close = threading.Event() - notifications = [] - errors: list[BaseException] = [] + session.process = _FakeProcess() def notify(method, params, **kwargs): - notifications.append(method) if method == "textDocument/didClose": - close_entered.set() - assert release_close.wait(timeout=2) - raise lsp.LspProtocolError("didClose write failed") + raise TimeoutError("didClose write timed out") monkeypatch.setattr(session, "_send_notification", notify) monkeypatch.setattr(session, "_collect_diagnostics", lambda uri, timeout: []) monkeypatch.setattr( session, "_send_request", - lambda method, params, timeout=30: { - "contents": {"kind": "plaintext", "value": "Nat : Type"} - }, + lambda method, params, timeout=30: {"contents": "Nat : Type"}, ) - def run() -> None: - try: - if operation == "diagnostics": - session.get_diagnostics(str(source)) - else: - session.hover(str(source), 0, 0) - except BaseException as error: - errors.append(error) - - first = threading.Thread(target=run) - second = threading.Thread(target=run) - first.start() - assert close_entered.wait(timeout=1) - second.start() - release_close.set() - first.join(timeout=2) - second.join(timeout=2) + def call(): + if operation == "diagnostics": + return session.get_diagnostics(str(source)) + return session.hover(str(source), 0, 0) - assert not first.is_alive() - assert not second.is_alive() - assert len(errors) == 2 - assert any("didClose write failed" in str(error) for error in errors) - assert any("session is retiring" in str(error) for error in errors) - assert notifications == ["textDocument/didOpen", "textDocument/didClose"] + assert call() == ([] if operation == "diagnostics" else "Nat : Type") + assert not session.is_alive() + with pytest.raises(lsp.LspProtocolError, match="retiring"): + call() def test_document_close_budget_exhaustion_poisons_the_session(tmp_path, monkeypatch): @@ -567,97 +257,14 @@ def finish_at_deadline(uri, timeout): monkeypatch.setattr(session, "_collect_diagnostics", finish_at_deadline) - with pytest.raises(TimeoutError, match="before closing the document"): - session.get_diagnostics(str(source)) - + assert session.get_diagnostics(str(source)) == [] assert notifications == ["textDocument/didOpen"] assert session._poisoned is True - assert session._retire_pending is True - - -@pytest.mark.parametrize("operation", ["diagnostics", "hover"]) -def test_did_close_cancellation_outranks_an_ordinary_operation_error( - tmp_path, monkeypatch, operation -): - source = tmp_path / "Test.lean" - source.write_text("#check Nat\n") - session = lsp.LeanLspSession(lsp.LspConfig()) - operation_error = lsp.LspProtocolError("operation failed") - close_cancellation = _Cancellation("cancel didClose") - - def notify(method, params, **kwargs): - if method == "textDocument/didClose": - raise close_cancellation - - monkeypatch.setattr(session, "_send_notification", notify) - monkeypatch.setattr( - session, - "_collect_diagnostics", - lambda uri, timeout: (_ for _ in ()).throw(operation_error), - ) - monkeypatch.setattr( - session, - "_send_request", - lambda method, params, timeout=30: (_ for _ in ()).throw(operation_error), - ) - - with pytest.raises(_Cancellation, match="cancel didClose") as raised: - if operation == "diagnostics": - session.get_diagnostics(str(source)) - else: - session.hover(str(source), 0, 0) - - assert raised.value is close_cancellation - assert raised.value.__cause__ is operation_error - assert session._poisoned is True - - -@pytest.mark.parametrize("operation", ["diagnostics", "hover"]) -def test_operation_cancellation_outranks_an_ordinary_did_close_error( - tmp_path, monkeypatch, operation -): - source = tmp_path / "Test.lean" - source.write_text("#check Nat\n") - session = lsp.LeanLspSession(lsp.LspConfig()) - operation_cancellation = _Cancellation("cancel operation") - close_error = lsp.LspProtocolError("didClose failed") - - def notify(method, params, **kwargs): - if method == "textDocument/didClose": - raise close_error - - monkeypatch.setattr(session, "_send_notification", notify) - monkeypatch.setattr( - session, - "_collect_diagnostics", - lambda uri, timeout: (_ for _ in ()).throw(operation_cancellation), - ) - monkeypatch.setattr( - session, - "_send_request", - lambda method, params, timeout=30: (_ for _ in ()).throw( - operation_cancellation - ), - ) - - with pytest.raises(_Cancellation, match="cancel operation") as raised: - if operation == "diagnostics": - session.get_diagnostics(str(source)) - else: - session.hover(str(source), 0, 0) - - assert raised.value is operation_cancellation - assert raised.value.__cause__ is close_error - assert session._poisoned is True def test_concurrent_abort_and_close_serialize_owned_cleanup(monkeypatch): process = _FakeProcess() - session = lsp.LeanLspSession(lsp.LspConfig()) - session.process = process - session._process_group_id = process.pid - session._group_retired = False - session._retire_pending = True + session = _owned_session(process) entered = threading.Event() release = threading.Event() active = 0 @@ -672,7 +279,7 @@ def retire(owned, process_group_id): maximum_active = max(maximum_active, active) entered.set() assert release.wait(timeout=2) - _finish_fake_process(owned) + _retire_fake_group(owned, process_group_id) with state_lock: active -= 1 @@ -689,6 +296,7 @@ def call(operation) -> None: aborter.start() assert entered.wait(timeout=1) closer.start() + closer.join(timeout=0.1) assert closer.is_alive() release.set() aborter.join(timeout=2) @@ -696,7 +304,7 @@ def call(operation) -> None: assert errors == [] assert maximum_active == 1 - assert session.is_clean() + assert session.process is None @pytest.mark.skipif(os.name != "posix", reason="native LSP backend requires POSIX groups") @@ -710,25 +318,20 @@ def test_close_retires_a_descendant_after_the_lsp_wrapper_exits(): [sys.executable, "-c", script], stdin=subprocess.PIPE, stdout=subprocess.PIPE, - stderr=subprocess.PIPE, text=True, start_new_session=True, ) assert process.stdout is not None child_pid = int(process.stdout.readline()) assert process.wait(timeout=5) == 0 - session = lsp.LeanLspSession(lsp.LspConfig()) - session.process = process - session._process_group_id = process.pid - session._group_retired = False + session = _owned_session(process) try: session.close() - assert session.is_clean() + assert session.process is None try: import psutil - child = psutil.Process(child_pid) - assert child.status() == psutil.STATUS_ZOMBIE + assert psutil.Process(child_pid).status() == psutil.STATUS_ZOMBIE except psutil.NoSuchProcess: pass finally: diff --git a/tests/test_shared_lean_runtime.py b/tests/test_shared_lean_runtime.py index 8e545a374..320747e2b 100644 --- a/tests/test_shared_lean_runtime.py +++ b/tests/test_shared_lean_runtime.py @@ -666,24 +666,32 @@ def test_failed_lsp_cleanup_is_invalidated_and_settled_before_replacement( from servers.lsp import server as lsp project = make_lake_project(tmp_path, "lsp-cleanup-order") - source = project / "Main.lean" - source.write_text("#check Nat\n") + (project / "Main.lean").write_text("#check Nat\n") sessions = [] errors = [] results = [] events = [] cleanup_started = threading.Event() release_cleanup = threading.Event() - attempts = 0 services = None + class Process: + returncode = None + stdin = None + stdout = None + + def __init__(self, pid): + self.pid = pid + + def poll(self): + return self.returncode + class Session(lsp.LeanLspSession): def __init__(self, root): super().__init__(lsp.LspConfig(cwd=str(root))) self.number = len(sessions) + 1 - self.process = _RuntimeFakeProcess(50000 + self.number) + self.process = Process(50000 + self.number) self._process_group_id = self.process.pid - self._group_retired = False sessions.append(self) events.append(f"create:{self.number}") @@ -692,33 +700,11 @@ def get_diagnostics(self, file_path): raise lsp.LspProtocolError("broken shared stream") return [] - def close(self): - self.abort() - - class _RuntimeFakeProcess: - def __init__(self, pid): - self.pid = pid - self.returncode = None - self.stdin = None - self.stdout = None - self.stderr = None - - def poll(self): - return self.returncode - - def wait(self, timeout=None): - return self.returncode - - def kill(self): - self.returncode = -9 - def retire(process, process_group_id): - nonlocal attempts if process is sessions[0].process: - attempts += 1 resident = services.lsp_projects.stats()["resident"] - events.append(f"retire:{attempts}:valid={resident[0]['valid']}") - if attempts == 1: + events.append(f"retire:valid={resident[0]['valid']}") + if not cleanup_started.is_set(): cleanup_started.set() raise RuntimeError("transient cleanup failure") assert release_cleanup.wait(timeout=2) @@ -746,23 +732,21 @@ def dispatch(target): first.start() assert cleanup_started.wait(timeout=1) second.start() + second.join(timeout=0.2) + # The replacement waits while the failed session still owns its group. assert len(sessions) == 1 - assert not results release_cleanup.set() first.join(timeout=2) second.join(timeout=2) - assert not first.is_alive() - assert not second.is_alive() - assert len(errors) == 1 - assert isinstance(errors[0], lsp.LspProtocolError) - assert str(errors[0]) == "broken shared stream" + assert not first.is_alive() and not second.is_alive() + assert [str(error) for error in errors] == ["broken shared stream"] assert results == ["No diagnostics — file compiles cleanly."] assert events[:4] == [ "create:1", - "retire:1:valid=False", - "retire:2:valid=False", + "retire:valid=False", + "retire:valid=False", "create:2", ] finally: @@ -772,252 +756,6 @@ def dispatch(target): services.close() -def test_lsp_cancellation_invalidates_then_cleans_before_it_escapes(tmp_path): - project = make_lake_project(tmp_path, "lsp-cancellation") - source = project / "Main.lean" - source.write_text("#check Nat\n") - events = [] - services = None - - class Cancellation(BaseException): - pass - - class Session(FakeLsp): - def get_diagnostics(self, file_path): - raise Cancellation("cancel diagnostics") - - def abort(self): - resident = services.lsp_projects.stats()["resident"] - events.append(f"abort:valid={resident[0]['valid']}") - super().abort() - - services = LeanRuntimeServices( - runtime_config(), - repl_factory=FakePool, - lsp_factory=Session, - start_sweepers=False, - ) - try: - with pytest.raises(Cancellation, match="cancel diagnostics"): - services.dispatch( - "lsp.diagnostics", - {"project_dir": str(project), "file_path": str(source)}, - ) - assert events == ["abort:valid=False"] - finally: - services.close() - - -def test_lsp_operation_cancellation_remains_primary_after_cleanup_cancellation( - tmp_path, monkeypatch -): - from servers.lsp import server as lsp - - project = make_lake_project(tmp_path, "lsp-double-cancellation") - source = project / "Main.lean" - source.write_text("#check Nat\n") - - class OperationCancellation(BaseException): - pass - - class CleanupCancellation(BaseException): - pass - - operation_error = OperationCancellation("cancel operation") - cleanup_error = CleanupCancellation("cancel cleanup") - attempts = 0 - sessions = [] - - class Process: - pid = 51000 - returncode = None - stdin = None - stdout = None - stderr = None - - def poll(self): - return self.returncode - - def wait(self, timeout=None): - return self.returncode - - class Session(lsp.LeanLspSession): - def __init__(self, root): - super().__init__(lsp.LspConfig(cwd=str(root))) - self.process = Process() - self._process_group_id = self.process.pid - self._group_retired = False - sessions.append(self) - - def get_diagnostics(self, file_path): - raise operation_error - - def close(self): - self.abort() - - def retire(process, process_group_id): - nonlocal attempts - attempts += 1 - if attempts == 1: - raise cleanup_error - process.returncode = -9 - - monkeypatch.setattr(lsp, "_kill_subprocesses", retire) - services = LeanRuntimeServices( - runtime_config(), - repl_factory=FakePool, - lsp_factory=Session, - start_sweepers=False, - ) - try: - with pytest.raises(OperationCancellation, match="cancel operation") as raised: - services.dispatch( - "lsp.diagnostics", - {"project_dir": str(project), "file_path": str(source)}, - ) - assert raised.value is operation_error - assert raised.value.__cause__ is cleanup_error - assert attempts == 2 - assert sessions[0].is_clean() - finally: - services.close() - - -def test_runtime_close_cleans_every_lsp_before_reraising_cancellation(tmp_path): - first = make_lake_project(tmp_path, "first-lsp-close") - second = make_lake_project(tmp_path, "second-lsp-close") - cancellation = _RuntimeCloseCancellation("cancel first close") - sessions = [] - events = [] - - class Session(FakeLsp): - def __init__(self, root): - super().__init__(root) - self.number = len(sessions) + 1 - self.close_calls = 0 - sessions.append(self) - - def get_diagnostics(self, file_path): - return [] - - def close(self): - self.close_calls += 1 - self.closed = True - events.append(f"close:{self.number}") - if self.number == 1 and self.close_calls == 1: - raise cancellation - - services = LeanRuntimeServices( - runtime_config(), - repl_factory=FakePool, - lsp_factory=Session, - start_sweepers=False, - ) - for project in (first, second): - source = project / "Main.lean" - source.write_text("#check Nat\n") - services.dispatch( - "lsp.diagnostics", - {"project_dir": str(project), "file_path": str(source)}, - ) - - with pytest.raises(_RuntimeCloseCancellation, match="cancel first close"): - services.close() - - assert events == ["close:1", "close:1", "close:2"] - assert all(session.closed for session in sessions) - - -def test_runtime_attempts_lsp_cleanup_after_repl_cleanup_cancellation(tmp_path): - project = make_lake_project(tmp_path, "repl-close-cancellation") - source = project / "Main.lean" - source.write_text("#check Nat\n") - cancellation = _RuntimeCloseCancellation("cancel REPL close") - pools = [] - sessions = [] - - class Pool(FakePool): - def __init__(self, root): - super().__init__(root) - self.shutdown_calls = 0 - - def shutdown(self): - self.shutdown_calls += 1 - self._shutdown = True - if self.shutdown_calls == 1: - raise cancellation - - class Session(FakeLsp): - def get_diagnostics(self, file_path): - return [] - - services = LeanRuntimeServices( - runtime_config(), - repl_factory=lambda root: pools.append(Pool(root)) or pools[-1], - lsp_factory=lambda root: sessions.append(Session(root)) or sessions[-1], - start_sweepers=False, - ) - services.dispatch( - "repl.run", - {"project_dir": str(project), "code": "#check Nat", "timeout": None}, - ) - services.dispatch( - "lsp.diagnostics", - {"project_dir": str(project), "file_path": str(source)}, - ) - - with pytest.raises(_RuntimeCloseCancellation, match="cancel REPL close"): - services.close() - - assert pools[0]._shutdown is True - assert pools[0].shutdown_calls == 2 - assert sessions[0].closed is True - - -def test_stale_resource_close_cancellation_releases_creation_reservation(tmp_path): - project = make_lake_project(tmp_path, "stale-close-cancellation") - cancellation = _RuntimeCloseCancellation("cancel stale close") - resources = [] - - class Resource: - def __init__(self): - self.valid = True - self.close_calls = 0 - self.interrupt_close = not resources - resources.append(self) - - def close(self): - self.close_calls += 1 - self.valid = False - if self.interrupt_close and self.close_calls == 1: - raise cancellation - - cache = ProjectResourceCache( - lambda root: Resource(), - lambda resource: resource.close(), - max_entries=1, - idle_seconds=1800, - is_valid=lambda resource: resource.valid, - start_sweeper=False, - ) - with cache.lease(str(project)) as first: - first.valid = False - - with pytest.raises(_RuntimeCloseCancellation, match="cancel stale close"): - with cache.lease(str(project)): - pytest.fail("replacement must wait for the close cancellation to escape") - - assert cache.state(str(project)) == "cold" - assert resources[0].close_calls == 2 - with cache.lease(str(project)) as replacement: - assert replacement is resources[1] - cache.close() - - -class _RuntimeCloseCancellation(BaseException): - pass - - @pytest.mark.skipif(not hasattr(os, "getuid"), reason="shared runtime requires POSIX locks") def test_runtime_holds_its_lifetime_lock_through_lsp_cleanup(runtime_dir, monkeypatch): import fcntl From ee352e598201ef27d3e25b3a30b02b383d6cf689 Mon Sep 17 00:00:00 2001 From: Jack McCarthy Date: Mon, 5 Oct 2026 19:01:38 -0400 Subject: [PATCH 04/10] Fit new LSP test signatures to the repo line length --- tests/test_lsp_protocol.py | 4 +--- tests/test_shared_lean_runtime.py | 4 +--- 2 files changed, 2 insertions(+), 6 deletions(-) diff --git a/tests/test_lsp_protocol.py b/tests/test_lsp_protocol.py index 89ddee8ac..3dc1c2f44 100644 --- a/tests/test_lsp_protocol.py +++ b/tests/test_lsp_protocol.py @@ -207,9 +207,7 @@ def run() -> None: @pytest.mark.parametrize("operation", ["diagnostics", "hover"]) -def test_did_close_failure_keeps_the_result_but_poisons_the_session( - tmp_path, monkeypatch, operation -): +def test_did_close_failure_keeps_the_result_but_poisons_the_session(tmp_path, monkeypatch, operation): source = tmp_path / "Test.lean" source.write_text("#check Nat\n") session = lsp.LeanLspSession(lsp.LspConfig()) diff --git a/tests/test_shared_lean_runtime.py b/tests/test_shared_lean_runtime.py index 320747e2b..570b38504 100644 --- a/tests/test_shared_lean_runtime.py +++ b/tests/test_shared_lean_runtime.py @@ -660,9 +660,7 @@ def get_diagnostics(self, file_path): services.close() -def test_failed_lsp_cleanup_is_invalidated_and_settled_before_replacement( - tmp_path, monkeypatch -): +def test_failed_lsp_cleanup_is_invalidated_and_settled_before_replacement(tmp_path, monkeypatch): from servers.lsp import server as lsp project = make_lake_project(tmp_path, "lsp-cleanup-order") From f793caeea6800c1ddf4d3d8181e1954123e543d9 Mon Sep 17 00:00:00 2001 From: Jack McCarthy Date: Mon, 5 Oct 2026 21:16:46 -0400 Subject: [PATCH 05/10] Keep a warm LSP session alive when a hover input file cannot be decoded hover() read the file inside the poisoning try, so a .lean file that is not valid UTF-8 poisoned the session, and the runtime's widened `except Exception` then invalidated it and killed its process group, failing any caller queued behind it. Read the file before admission, and let the runtime re-raise ValueError (which includes UnicodeDecodeError) without retiring the session. Stream decode errors are still wrapped as LspProtocolError, so this route cannot spare a broken stream. --- servers/lean_runtime.py | 6 ++-- servers/lsp/server.py | 10 ++++--- tests/test_shared_lean_runtime.py | 49 +++++++++++++++++++++++++++++++ 3 files changed, 59 insertions(+), 6 deletions(-) diff --git a/servers/lean_runtime.py b/servers/lean_runtime.py index c20552815..b8c9e4398 100644 --- a/servers/lean_runtime.py +++ b/servers/lean_runtime.py @@ -743,7 +743,8 @@ def dispatch(self, method: str, params: dict[str, Any]) -> Any: assert session is not None try: diagnostics = session.get_diagnostics(str(path)) - except LspBusyError: + except (LspBusyError, ValueError): + # A queue timeout or undecodable input file leaves the session healthy. raise except Exception: self.lsp_projects.invalidate(str(root), session) @@ -766,7 +767,8 @@ def dispatch(self, method: str, params: dict[str, Any]) -> Any: assert session is not None try: result = session.hover(str(path), line, character) - except LspBusyError: + except (LspBusyError, ValueError): + # A queue timeout or undecodable input file leaves the session healthy. raise except Exception: self.lsp_projects.invalidate(str(root), session) diff --git a/servers/lsp/server.py b/servers/lsp/server.py index 0c9c4d9c9..a0aedaca5 100644 --- a/servers/lsp/server.py +++ b/servers/lsp/server.py @@ -227,6 +227,9 @@ def _get_diagnostics(self, file_path: str, *, timeout: float | None = None) -> l def hover(self, file_path: str, line: int, character: int) -> str | None: """Get hover information at a position.""" + # Read before admission so a bad input file cannot poison the session. + path = Path(file_path).resolve() + content = path.read_text() deadline = time.monotonic() + self.config.timeout if not self._operation_lock.acquire(timeout=self.config.timeout): raise LspBusyError( @@ -241,7 +244,7 @@ def hover(self, file_path: str, line: int, character: int) -> str | None: f"timed out after {self.config.timeout:g}s waiting for the Lean LSP session" ) try: - return self._hover(file_path, line, character, timeout=remaining) + return self._hover(path, content, line, character, timeout=remaining) except BaseException: # Poison before releasing admission so a queued call cannot # read the failed operation's leftovers from the stream. @@ -252,14 +255,13 @@ def hover(self, file_path: str, line: int, character: int) -> str | None: def _hover( self, - file_path: str, + path: Path, + content: str, line: int, character: int, *, timeout: float = 30, ) -> str | None: - path = Path(file_path).resolve() - content = path.read_text() uri = path.as_uri() deadline = time.monotonic() + timeout self._send_notification( diff --git a/tests/test_shared_lean_runtime.py b/tests/test_shared_lean_runtime.py index 570b38504..e0668c235 100644 --- a/tests/test_shared_lean_runtime.py +++ b/tests/test_shared_lean_runtime.py @@ -660,6 +660,55 @@ def get_diagnostics(self, file_path): services.close() +def test_unreadable_hover_file_leaves_a_warm_lsp_session_alone(tmp_path): + from servers.lsp import server as lsp + + project = make_lake_project(tmp_path, "lsp-input") + (project / "Main.lean").write_text("#check Nat\n") + (project / "Latin1.lean").write_bytes("-- café\n".encode("latin-1")) + sessions = [] + + class Process: + stdin = None + stdout = None + + def poll(self): + return None + + class Session(lsp.LeanLspSession): + def __init__(self, root): + super().__init__(lsp.LspConfig(cwd=str(root))) + self.process = Process() + sessions.append(self) + + def _send_notification(self, method, params, **kwargs): + pass + + def _send_request(self, method, params, timeout=30): + return {"contents": "Nat : Type"} + + services = LeanRuntimeServices( + runtime_config(), + repl_factory=FakePool, + lsp_factory=Session, + start_sweepers=False, + ) + + def hover(file_path): + params = {"project_dir": str(project), "file_path": file_path, "line": 0, "character": 0} + return services.dispatch("lsp.hover", params) + + try: + assert hover("Main.lean") == "Nat : Type" + with pytest.raises(UnicodeDecodeError): + hover("Latin1.lean") + assert sessions[0].is_alive() + assert hover("Main.lean") == "Nat : Type" + assert len(sessions) == 1 + finally: + services.close() + + def test_failed_lsp_cleanup_is_invalidated_and_settled_before_replacement(tmp_path, monkeypatch): from servers.lsp import server as lsp From bb5048261a11f348adcb465c9619b50abea2fd5f Mon Sep 17 00:00:00 2001 From: Jack McCarthy Date: Mon, 5 Oct 2026 21:19:49 -0400 Subject: [PATCH 06/10] Pin that a busy LSP session is neither retired nor replaced --- tests/test_shared_lean_runtime.py | 42 +++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/tests/test_shared_lean_runtime.py b/tests/test_shared_lean_runtime.py index e0668c235..92d38a941 100644 --- a/tests/test_shared_lean_runtime.py +++ b/tests/test_shared_lean_runtime.py @@ -660,6 +660,48 @@ def get_diagnostics(self, file_path): services.close() +def test_busy_lsp_session_is_left_alone_and_reused(tmp_path): + from servers.lsp.server import LspBusyError + + project = make_lake_project(tmp_path, "lsp-busy") + (project / "Main.lean").write_text("#check Nat\n") + sessions = [] + + class Session(FakeLsp): + def __init__(self, root): + super().__init__(root) + self.busy = True + self.aborted = False + sessions.append(self) + + def abort(self): + self.aborted = True + super().abort() + + def get_diagnostics(self, file_path): + if self.busy: + self.busy = False + raise LspBusyError("timed out waiting for the Lean LSP session") + return [] + + services = LeanRuntimeServices( + runtime_config(), + repl_factory=FakePool, + lsp_factory=Session, + start_sweepers=False, + ) + try: + params = {"project_dir": str(project), "file_path": "Main.lean"} + with pytest.raises(LspBusyError): + services.dispatch("lsp.diagnostics", params) + assert services.lsp_projects.stats()["resident"][0]["valid"] is True + assert sessions[0].aborted is False + assert services.dispatch("lsp.diagnostics", params).startswith("No diagnostics") + assert len(sessions) == 1 + finally: + services.close() + + def test_unreadable_hover_file_leaves_a_warm_lsp_session_alone(tmp_path): from servers.lsp import server as lsp From 58ff1a2e23aa67b8fd522e64efe9b80bec732115 Mon Sep 17 00:00:00 2001 From: Jack McCarthy Date: Mon, 5 Oct 2026 21:29:12 -0400 Subject: [PATCH 07/10] Reuse _remaining for the didClose budget checks --- servers/lsp/server.py | 28 ++++++++++------------------ 1 file changed, 10 insertions(+), 18 deletions(-) diff --git a/servers/lsp/server.py b/servers/lsp/server.py index a0aedaca5..771e823f0 100644 --- a/servers/lsp/server.py +++ b/servers/lsp/server.py @@ -210,15 +210,11 @@ def _get_diagnostics(self, file_path: str, *, timeout: float | None = None) -> l ) finally: try: - remaining = deadline - time.monotonic() - if remaining > 0: - self._send_notification( - "textDocument/didClose", - {"textDocument": {"uri": uri}}, - timeout=remaining, - ) - else: - raise TimeoutError("no time remained to close the document") + self._send_notification( + "textDocument/didClose", + {"textDocument": {"uri": uri}}, + timeout=self._remaining(deadline, operation_timeout), + ) except Exception: # An unclosed document or a partial frame desynchronizes the # stream; keep this result but never reuse the session. @@ -287,15 +283,11 @@ def _hover( ) finally: try: - remaining = deadline - time.monotonic() - if remaining > 0: - self._send_notification( - "textDocument/didClose", - {"textDocument": {"uri": uri}}, - timeout=remaining, - ) - else: - raise TimeoutError("no time remained to close the document") + self._send_notification( + "textDocument/didClose", + {"textDocument": {"uri": uri}}, + timeout=self._remaining(deadline, timeout), + ) except Exception: # An unclosed document or a partial frame desynchronizes the # stream; keep this result but never reuse the session. From 7c378319112988c39ae01c86737b7ab4222a225b Mon Sep 17 00:00:00 2001 From: Jack McCarthy <37917934+Deicyde@users.noreply.github.com> Date: Tue, 6 Oct 2026 02:52:54 -0400 Subject: [PATCH 08/10] Preserve healthy LSP sessions on input failures --- servers/lean_runtime.py | 62 +++++------ servers/lsp/server.py | 47 ++++----- tests/test_lsp_protocol.py | 45 ++++++++ tests/test_shared_lean_runtime.py | 168 ++++++++++++++++++++++++++++++ 4 files changed, 265 insertions(+), 57 deletions(-) diff --git a/servers/lean_runtime.py b/servers/lean_runtime.py index b8c9e4398..dbcb0e18d 100644 --- a/servers/lean_runtime.py +++ b/servers/lean_runtime.py @@ -41,7 +41,6 @@ ) from servers.lsp.server import ( LspConfig, - LspBusyError, LeanLspSession, format_lsp_diagnostics, ) @@ -735,21 +734,10 @@ def dispatch(self, method: str, params: dict[str, Any]) -> Any: project_dir = self._string_param(params, "project_dir") file_path = self._string_param(params, "file_path") root, path = resolve_lean_file(project_dir, file_path) - with self.lsp_projects.lease( - str(root), - acquisition_timeout=self._acquisition_timeout(self.config.lsp_timeout), - creation_budget=self.lsp_creation_budget, - ) as session: - assert session is not None - try: - diagnostics = session.get_diagnostics(str(path)) - except (LspBusyError, ValueError): - # A queue timeout or undecodable input file leaves the session healthy. - raise - except Exception: - self.lsp_projects.invalidate(str(root), session) - session.abort() - raise + diagnostics = self._lsp_operation( + root, + lambda session: session.get_diagnostics(str(path)), + ) return format_lsp_diagnostics(diagnostics) if method == "lsp.hover": project_dir = self._string_param(params, "project_dir") @@ -759,24 +747,36 @@ def dispatch(self, method: str, params: dict[str, Any]) -> Any: if line < 0 or character < 0: raise ValueError("line and character must be nonnegative") root, path = resolve_lean_file(project_dir, file_path) - with self.lsp_projects.lease( - str(root), - acquisition_timeout=self._acquisition_timeout(self.config.lsp_timeout), - creation_budget=self.lsp_creation_budget, - ) as session: - assert session is not None - try: - result = session.hover(str(path), line, character) - except (LspBusyError, ValueError): - # A queue timeout or undecodable input file leaves the session healthy. - raise - except Exception: - self.lsp_projects.invalidate(str(root), session) - session.abort() - raise + result = self._lsp_operation( + root, + lambda session: session.hover(str(path), line, character), + ) return result or "No hover information at this position." raise ValueError(f"unknown Lean runtime method: {method}") + def _lsp_operation( + self, + root: Path, + operation: Callable[[LeanLspSession], T], + ) -> T: + """Route one LSP call without retiring a healthy shared session.""" + with self.lsp_projects.lease( + str(root), + acquisition_timeout=self._acquisition_timeout(self.config.lsp_timeout), + creation_budget=self.lsp_creation_budget, + ) as session: + assert session is not None + try: + return operation(session) + except Exception: + # Real operations poison the session before releasing admission + # whenever the shared stream may be desynchronized. Input reads + # and queue timeouts happen before that point and leave it alive. + if not session.is_alive(): + self.lsp_projects.invalidate(str(root), session) + session.abort() + raise + def status(self, *, include_projects: bool) -> dict[str, Any]: result: dict[str, Any] = { "running": True, diff --git a/servers/lsp/server.py b/servers/lsp/server.py index 771e823f0..04a2224c0 100644 --- a/servers/lsp/server.py +++ b/servers/lsp/server.py @@ -15,7 +15,7 @@ from dataclasses import dataclass, field from logging import getLogger from pathlib import Path -from typing import Any +from typing import Any, TypeVar from fastmcp.server import FastMCP @@ -25,6 +25,8 @@ logger = getLogger(__name__) +T = TypeVar("T") + DEFAULT_LSP_TIMEOUT = 60 MAX_LSP_HEADER_BYTES = 16 * 1024 MAX_LSP_MESSAGE_BYTES = 16 * 1024 * 1024 @@ -151,8 +153,8 @@ def is_alive(self) -> bool: and self.process.poll() is None ) - def get_diagnostics(self, file_path: str) -> list[dict]: - """Open a file and collect diagnostics from the language server.""" + def _admitted(self, operation: Callable[[float], T]) -> T: + """Run one serialized operation without poisoning on admission failure.""" deadline = time.monotonic() + self.config.timeout if not self._operation_lock.acquire(timeout=self.config.timeout): raise LspBusyError( @@ -167,7 +169,7 @@ def get_diagnostics(self, file_path: str) -> list[dict]: f"timed out after {self.config.timeout:g}s waiting for the Lean LSP session" ) try: - return self._get_diagnostics(file_path, timeout=remaining) + return operation(remaining) except BaseException: # Poison before releasing admission so a queued call cannot # read the failed operation's leftovers from the stream. @@ -176,6 +178,12 @@ def get_diagnostics(self, file_path: str) -> list[dict]: finally: self._operation_lock.release() + def get_diagnostics(self, file_path: str) -> list[dict]: + """Open a file and collect diagnostics from the language server.""" + return self._admitted( + lambda remaining: self._get_diagnostics(file_path, timeout=remaining) + ) + def _get_diagnostics(self, file_path: str, *, timeout: float | None = None) -> list[dict]: path = Path(file_path).resolve() uri = path.as_uri() @@ -226,28 +234,15 @@ def hover(self, file_path: str, line: int, character: int) -> str | None: # Read before admission so a bad input file cannot poison the session. path = Path(file_path).resolve() content = path.read_text() - deadline = time.monotonic() + self.config.timeout - if not self._operation_lock.acquire(timeout=self.config.timeout): - raise LspBusyError( - f"timed out after {self.config.timeout:g}s waiting for the Lean LSP session" + return self._admitted( + lambda remaining: self._hover( + path, + content, + line, + character, + timeout=remaining, ) - try: - if self._poisoned: - raise LspProtocolError("Lean LSP session is retiring after a failed operation") - remaining = deadline - time.monotonic() - if remaining <= 0: - raise LspBusyError( - f"timed out after {self.config.timeout:g}s waiting for the Lean LSP session" - ) - try: - return self._hover(path, content, line, character, timeout=remaining) - except BaseException: - # Poison before releasing admission so a queued call cannot - # read the failed operation's leftovers from the stream. - self._poisoned = True - raise - finally: - self._operation_lock.release() + ) def _hover( self, @@ -486,7 +481,7 @@ def _read_message(self, timeout: float = 5) -> dict | None: try: message = json.loads(bytes(body).decode("utf-8")) - except (UnicodeDecodeError, json.JSONDecodeError) as error: + except ValueError as error: raise LspProtocolError("LSP emitted an invalid JSON body") from error if not isinstance(message, dict): raise LspProtocolError("LSP JSON-RPC message is not an object") diff --git a/tests/test_lsp_protocol.py b/tests/test_lsp_protocol.py index 3dc1c2f44..cc64a184a 100644 --- a/tests/test_lsp_protocol.py +++ b/tests/test_lsp_protocol.py @@ -206,6 +206,35 @@ def run() -> None: ] +@pytest.mark.parametrize("operation", ["diagnostics", "hover"]) +def test_post_admission_budget_exhaustion_keeps_the_session_healthy( + tmp_path, monkeypatch, operation +): + source = tmp_path / "Test.lean" + source.write_text("#check Nat\n") + session = lsp.LeanLspSession(lsp.LspConfig(timeout=0)) + session.process = _FakeProcess() + monkeypatch.setattr(lsp.time, "monotonic", lambda: 0.0) + monkeypatch.setattr( + session, + "_get_diagnostics", + lambda *args, **kwargs: pytest.fail("expired admission ran diagnostics"), + ) + monkeypatch.setattr( + session, + "_hover", + lambda *args, **kwargs: pytest.fail("expired admission ran hover"), + ) + + with pytest.raises(lsp.LspBusyError, match="waiting for the Lean LSP session"): + if operation == "diagnostics": + session.get_diagnostics(str(source)) + else: + session.hover(str(source), 0, 0) + + assert session.is_alive() + + @pytest.mark.parametrize("operation", ["diagnostics", "hover"]) def test_did_close_failure_keeps_the_result_but_poisons_the_session(tmp_path, monkeypatch, operation): source = tmp_path / "Test.lean" @@ -513,6 +542,22 @@ def test_lsp_queue_wait_is_bounded_by_session_timeout(tmp_path: Path): session._operation_lock.release() +def test_lsp_json_integer_limit_is_reported_as_a_protocol_error(): + body = b'{"jsonrpc":"2.0","id":' + (b"9" * 5000) + b"}" + frame = f"Content-Length: {len(body)}\r\n\r\n".encode("ascii") + body + read_fd, write_fd = os.pipe() + reader = os.fdopen(read_fd, "rb", buffering=0) + session = lsp.LeanLspSession(lsp.LspConfig()) + session.process = SimpleNamespace(stdout=reader) + try: + os.write(write_fd, frame) + with pytest.raises(lsp.LspProtocolError, match="invalid JSON body"): + session._read_message(timeout=1) + finally: + os.close(write_fd) + reader.close() + + @pytest.mark.parametrize( "partial", [ diff --git a/tests/test_shared_lean_runtime.py b/tests/test_shared_lean_runtime.py index 92d38a941..c9cf8e0d9 100644 --- a/tests/test_shared_lean_runtime.py +++ b/tests/test_shared_lean_runtime.py @@ -9,6 +9,7 @@ import sys import threading import time +from pathlib import Path import pytest @@ -702,6 +703,67 @@ def get_diagnostics(self, file_path): services.close() +def test_hover_busy_after_admission_leaves_the_session_alive(tmp_path): + from servers.lsp import server as lsp + + project = make_lake_project(tmp_path, "lsp-hover-busy") + (project / "Main.lean").write_text("#check Nat\n") + sessions = [] + + class Process: + stdin = None + stdout = None + returncode = None + + def poll(self): + return self.returncode + + def wait(self, timeout=None): + self.returncode = 0 + return self.returncode + + class Session(lsp.LeanLspSession): + def __init__(self, root): + super().__init__(lsp.LspConfig(cwd=str(root), timeout=0)) + self.process = Process() + self.aborted = False + sessions.append(self) + + def abort(self): + self.aborted = True + super().abort() + + def _send_notification(self, method, params, **kwargs): + pass + + def _send_request(self, method, params, timeout=30): + return {"contents": "Nat : Type"} + + services = LeanRuntimeServices( + runtime_config(), + repl_factory=FakePool, + lsp_factory=Session, + start_sweepers=False, + ) + params = { + "project_dir": str(project), + "file_path": "Main.lean", + "line": 0, + "character": 0, + } + try: + with pytest.raises(lsp.LspBusyError): + services.dispatch("lsp.hover", params) + assert services.lsp_projects.stats()["resident"][0]["valid"] is True + assert sessions[0].aborted is False + + sessions[0].config.timeout = 1 + assert services.dispatch("lsp.hover", params) == "Nat : Type" + assert len(sessions) == 1 + finally: + services.close() + + def test_unreadable_hover_file_leaves_a_warm_lsp_session_alone(tmp_path): from servers.lsp import server as lsp @@ -751,6 +813,109 @@ def hover(file_path): services.close() +def test_permission_error_hover_does_not_abort_a_concurrent_lsp_request( + tmp_path, + monkeypatch, +): + from servers.lsp import server as lsp + + project = make_lake_project(tmp_path, "lsp-concurrent-input") + readable = project / "Main.lean" + unreadable = project / "Unreadable.lean" + readable.write_text("#check Nat\n") + unreadable.write_text("#check Int\n") + diagnostics_started = threading.Event() + release_diagnostics = threading.Event() + sessions = [] + results = [] + errors = [] + + class Process: + stdin = None + stdout = None + returncode = None + + def poll(self): + return self.returncode + + def wait(self, timeout=None): + self.returncode = 0 + return self.returncode + + class Session(lsp.LeanLspSession): + def __init__(self, root): + super().__init__(lsp.LspConfig(cwd=str(root))) + self.process = Process() + self.aborted = False + sessions.append(self) + + def abort(self): + self.aborted = True + super().abort() + + def _get_diagnostics(self, file_path, *, timeout=None): + diagnostics_started.set() + assert release_diagnostics.wait(timeout=2) + return [] + + def _send_notification(self, method, params, **kwargs): + pass + + def _send_request(self, method, params, timeout=30): + return {"contents": "Nat : Type"} + + original_read_text = Path.read_text + + def read_text(path, *args, **kwargs): + if path.resolve() == unreadable.resolve(): + raise PermissionError("permission denied") + return original_read_text(path, *args, **kwargs) + + monkeypatch.setattr(Path, "read_text", read_text) + services = LeanRuntimeServices( + runtime_config(), + repl_factory=FakePool, + lsp_factory=Session, + start_sweepers=False, + ) + diagnostic_params = {"project_dir": str(project), "file_path": "Main.lean"} + hover_params = { + "project_dir": str(project), + "file_path": "Unreadable.lean", + "line": 0, + "character": 0, + } + + def diagnose(): + try: + results.append(services.dispatch("lsp.diagnostics", diagnostic_params)) + except BaseException as error: + errors.append(error) + + thread = threading.Thread(target=diagnose) + try: + thread.start() + assert diagnostics_started.wait(timeout=1) + with pytest.raises(PermissionError, match="permission denied"): + services.dispatch("lsp.hover", hover_params) + + assert thread.is_alive() + assert sessions[0].is_alive() + assert sessions[0].aborted is False + assert services.lsp_projects.stats()["resident"][0]["valid"] is True + + release_diagnostics.set() + thread.join(timeout=2) + assert not thread.is_alive() + assert errors == [] + assert results == ["No diagnostics — file compiles cleanly."] + assert len(sessions) == 1 + finally: + release_diagnostics.set() + thread.join(timeout=2) + services.close() + + def test_failed_lsp_cleanup_is_invalidated_and_settled_before_replacement(tmp_path, monkeypatch): from servers.lsp import server as lsp @@ -786,6 +951,9 @@ def __init__(self, root): def get_diagnostics(self, file_path): if self.number == 1: + # Real LeanLspSession operations poison before propagating a + # stream failure; keep this fake faithful to that contract. + self._poisoned = True raise lsp.LspProtocolError("broken shared stream") return [] From 2ce7ecb44a71277b675806df38aa9d678b71bb79 Mon Sep 17 00:00:00 2001 From: Jack McCarthy Date: Tue, 6 Oct 2026 08:37:38 -0400 Subject: [PATCH 09/10] Normalize and consolidate shared LSP calls --- servers/lean_runtime.py | 57 ++++++++++++++++--------------- servers/lsp/server.py | 50 +++++++++++---------------- tests/test_lsp_protocol.py | 70 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 119 insertions(+), 58 deletions(-) diff --git a/servers/lean_runtime.py b/servers/lean_runtime.py index beabbfaf7..402a04c1a 100644 --- a/servers/lean_runtime.py +++ b/servers/lean_runtime.py @@ -941,21 +941,10 @@ def dispatch(self, method: str, params: dict[str, Any]) -> Any: project_dir = self._string_param(params, "project_dir") file_path = self._string_param(params, "file_path") root, path = resolve_lean_file(project_dir, file_path) - with self.lsp_projects.lease_resolved( + diagnostics = self._lsp_operation( root, - acquisition_timeout=self._acquisition_timeout(self.config.lsp_timeout), - creation_budget=self.lsp_creation_budget, - ) as session: - assert session is not None - try: - diagnostics = session.get_diagnostics(str(path)) - except Exception: - if not session.is_alive(): - session.retire() - self.lsp_projects.invalidate_resolved(root, session) - raise - if not session.is_alive(): - self.lsp_projects.invalidate_resolved(root, session) + lambda session: session.get_diagnostics(str(path)), + ) return format_lsp_diagnostics(diagnostics) if method == "lsp.hover": project_dir = self._string_param(params, "project_dir") @@ -965,24 +954,36 @@ def dispatch(self, method: str, params: dict[str, Any]) -> Any: if line < 0 or character < 0: raise ValueError("line and character must be nonnegative") root, path = resolve_lean_file(project_dir, file_path) - with self.lsp_projects.lease_resolved( + result = self._lsp_operation( root, - acquisition_timeout=self._acquisition_timeout(self.config.lsp_timeout), - creation_budget=self.lsp_creation_budget, - ) as session: - assert session is not None - try: - result = session.hover(str(path), line, character) - except Exception: - if not session.is_alive(): - session.retire() - self.lsp_projects.invalidate_resolved(root, session) - raise - if not session.is_alive(): - self.lsp_projects.invalidate_resolved(root, session) + lambda session: session.hover(str(path), line, character), + ) return result or "No hover information at this position." raise ValueError(f"unknown Lean runtime method: {method}") + def _lsp_operation( + self, + root: Path, + operation: Callable[[LeanLspSession], T], + ) -> T: + """Route one LSP call while preserving retained cleanup ownership.""" + with self.lsp_projects.lease_resolved( + root, + acquisition_timeout=self._acquisition_timeout(self.config.lsp_timeout), + creation_budget=self.lsp_creation_budget, + ) as session: + assert session is not None + try: + result = operation(session) + except Exception: + if not session.is_alive(): + session.retire() + self.lsp_projects.invalidate_resolved(root, session) + raise + if not session.is_alive(): + self.lsp_projects.invalidate_resolved(root, session) + return result + def status(self, *, include_projects: bool) -> dict[str, Any]: result: dict[str, Any] = { "running": True, diff --git a/servers/lsp/server.py b/servers/lsp/server.py index 8518044e1..d2ef08be7 100644 --- a/servers/lsp/server.py +++ b/servers/lsp/server.py @@ -15,7 +15,7 @@ from dataclasses import dataclass, field from logging import getLogger from pathlib import Path -from typing import Any +from typing import Any, TypeVar from fastmcp.server import FastMCP @@ -30,6 +30,7 @@ logger = getLogger(__name__) _POPEN_TYPE = subprocess.Popen +T = TypeVar("T") DEFAULT_LSP_TIMEOUT = 60 MAX_LSP_HEADER_BYTES = 16 * 1024 @@ -214,8 +215,8 @@ def is_alive(self) -> bool: return True return status not in {psutil.STATUS_ZOMBIE, psutil.STATUS_DEAD} - def get_diagnostics(self, file_path: str) -> list[dict]: - """Open a file and collect diagnostics from the language server.""" + def _admitted(self, operation: Callable[[float], T]) -> T: + """Run one serialized operation and fence a failed protocol stream.""" deadline = time.monotonic() + self.config.timeout if not self._operation_lock.acquire(timeout=self.config.timeout): raise LspBusyError( @@ -230,13 +231,19 @@ def get_diagnostics(self, file_path: str) -> list[dict]: f"timed out after {self.config.timeout:g}s waiting for the Lean LSP session" ) try: - return self._get_diagnostics(file_path, timeout=remaining) + return operation(remaining) except BaseException: self._retire_pending = True raise finally: self._operation_lock.release() + def get_diagnostics(self, file_path: str) -> list[dict]: + """Open a file and collect diagnostics from the language server.""" + return self._admitted( + lambda remaining: self._get_diagnostics(file_path, timeout=remaining) + ) + def _get_diagnostics(self, file_path: str, *, timeout: float | None = None) -> list[dict]: path = Path(file_path).resolve() uri = path.as_uri() @@ -289,32 +296,15 @@ def hover(self, file_path: str, line: int, character: int) -> str | None: # shared protocol stream, so they must not retire a healthy session. path = Path(file_path).resolve() content = path.read_text() - deadline = time.monotonic() + self.config.timeout - if not self._operation_lock.acquire(timeout=self.config.timeout): - raise LspBusyError( - f"timed out after {self.config.timeout:g}s waiting for the Lean LSP session" + return self._admitted( + lambda remaining: self._hover( + path, + content, + line, + character, + timeout=remaining, ) - try: - if self._retire_pending: - raise LspProtocolError("Lean LSP session is retiring after a failed operation") - remaining = deadline - time.monotonic() - if remaining <= 0: - raise LspBusyError( - f"timed out after {self.config.timeout:g}s waiting for the Lean LSP session" - ) - try: - return self._hover( - path, - content, - line, - character, - timeout=remaining, - ) - except BaseException: - self._retire_pending = True - raise - finally: - self._operation_lock.release() + ) def _hover( self, @@ -554,7 +544,7 @@ def _read_message(self, timeout: float = 5) -> dict | None: try: message = json.loads(bytes(body).decode("utf-8")) - except (UnicodeDecodeError, json.JSONDecodeError) as error: + except ValueError as error: raise LspProtocolError("LSP emitted an invalid JSON body") from error if not isinstance(message, dict): raise LspProtocolError("LSP JSON-RPC message is not an object") diff --git a/tests/test_lsp_protocol.py b/tests/test_lsp_protocol.py index d0604baa5..4bac4b5df 100644 --- a/tests/test_lsp_protocol.py +++ b/tests/test_lsp_protocol.py @@ -340,6 +340,60 @@ def notify(method, params, **kwargs): session.hover(str(source), 0, 0) +@pytest.mark.parametrize("operation", ["diagnostics", "hover"]) +def test_post_admission_budget_exhaustion_keeps_session_healthy( + tmp_path, monkeypatch, operation +): + source = tmp_path / "Test.lean" + source.write_text("#check Nat\n") + session = _owned_session(_FakeProcess()) + session.config.timeout = 0 + monkeypatch.setattr(lsp.time, "monotonic", lambda: 0.0) + monkeypatch.setattr( + session, + "_get_diagnostics", + lambda *args, **kwargs: pytest.fail("expired admission ran diagnostics"), + ) + monkeypatch.setattr( + session, + "_hover", + lambda *args, **kwargs: pytest.fail("expired admission ran hover"), + ) + + with pytest.raises(lsp.LspBusyError, match="waiting for the Lean LSP session"): + if operation == "diagnostics": + session.get_diagnostics(str(source)) + else: + session.hover(str(source), 0, 0) + + assert session.is_alive() is True + assert session._retire_pending is False + + +def test_document_close_budget_exhaustion_retires_session(tmp_path, monkeypatch): + source = tmp_path / "Test.lean" + source.write_text("#check Nat\n") + session = lsp.LeanLspSession(lsp.LspConfig(timeout=60)) + clock = {"now": 0.0} + notifications = [] + monkeypatch.setattr(lsp.time, "monotonic", lambda: clock["now"]) + monkeypatch.setattr( + session, + "_send_notification", + lambda method, params, **kwargs: notifications.append(method), + ) + + def finish_at_deadline(uri, timeout): + clock["now"] = 60.0 + return [] + + monkeypatch.setattr(session, "_collect_diagnostics", finish_at_deadline) + + assert session.get_diagnostics(str(source)) == [] + assert notifications == ["textDocument/didOpen"] + assert session._retire_pending is True + + def test_hover_input_decode_failure_keeps_healthy_session(tmp_path): source = tmp_path / "Invalid.lean" source.write_bytes(b"\xff") @@ -644,6 +698,22 @@ def test_lsp_queue_wait_is_bounded_by_session_timeout(tmp_path: Path): session._operation_lock.release() +def test_lsp_json_integer_limit_is_reported_as_a_protocol_error(): + body = b'{"jsonrpc":"2.0","id":' + (b"9" * 5000) + b"}" + frame = f"Content-Length: {len(body)}\r\n\r\n".encode("ascii") + body + read_fd, write_fd = os.pipe() + reader = os.fdopen(read_fd, "rb", buffering=0) + session = lsp.LeanLspSession(lsp.LspConfig()) + session.process = SimpleNamespace(stdout=reader) + try: + os.write(write_fd, frame) + with pytest.raises(lsp.LspProtocolError, match="invalid JSON body"): + session._read_message(timeout=1) + finally: + os.close(write_fd) + reader.close() + + @pytest.mark.parametrize( "partial", [ From 0f789c2849651239e270dc1179d415b3b33e06ac Mon Sep 17 00:00:00 2001 From: Jack McCarthy Date: Sat, 10 Oct 2026 23:22:00 -0400 Subject: [PATCH 10/10] Report deeply nested LSP JSON as a protocol error json.loads raises RecursionError, not ValueError, when a body nests arrays or objects past the decoder's recursion limit. A 400 KB body of 200,000 nested arrays, far under MAX_LSP_MESSAGE_BYTES, therefore reached callers as a raw RecursionError instead of LspProtocolError. The session was already retired correctly; only the reported error was wrong. Catch RecursionError alongside ValueError and pin it with that body, which exceeds the limit on both Python 3.10 and 3.13. --- servers/lsp/server.py | 2 +- tests/test_lsp_protocol.py | 14 ++++++++++++++ 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/servers/lsp/server.py b/servers/lsp/server.py index d2ef08be7..5192dbb87 100644 --- a/servers/lsp/server.py +++ b/servers/lsp/server.py @@ -544,7 +544,7 @@ def _read_message(self, timeout: float = 5) -> dict | None: try: message = json.loads(bytes(body).decode("utf-8")) - except ValueError as error: + except (ValueError, RecursionError) as error: raise LspProtocolError("LSP emitted an invalid JSON body") from error if not isinstance(message, dict): raise LspProtocolError("LSP JSON-RPC message is not an object") diff --git a/tests/test_lsp_protocol.py b/tests/test_lsp_protocol.py index 4bac4b5df..b7647d020 100644 --- a/tests/test_lsp_protocol.py +++ b/tests/test_lsp_protocol.py @@ -714,6 +714,20 @@ def test_lsp_json_integer_limit_is_reported_as_a_protocol_error(): reader.close() +def test_lsp_json_nesting_limit_is_reported_as_a_protocol_error(tmp_path: Path): + # 200,000 levels overrun the JSON decoder's recursion limit (about 10,000 + # on Python 3.13, 1,000 on 3.10) in a body far under MAX_LSP_MESSAGE_BYTES. + # A pipe cannot buffer the whole frame, so the session reads it from a file. + body = b"[" * 200_000 + b"]" * 200_000 + frame = tmp_path / "frame" + frame.write_bytes(f"Content-Length: {len(body)}\r\n\r\n".encode("ascii") + body) + session = lsp.LeanLspSession(lsp.LspConfig()) + with frame.open("rb", buffering=0) as reader: + session.process = SimpleNamespace(stdout=reader) + with pytest.raises(lsp.LspProtocolError, match="invalid JSON body"): + session._read_message(timeout=1) + + @pytest.mark.parametrize( "partial", [