fix(lsp): reap idle language servers

agent/lsp/manager.py

@@ -176,6 +176,7 @@ class LSPService:
         self._spawning: Dict[Tuple[str, str], asyncio.Future] = {}
         self._last_used: Dict[Tuple[str, str], float] = {}
         self._state_lock = threading.Lock()
+        self._idle_reaper_task: Optional[asyncio.Task] = None
 
         # Delta baseline: file path → snapshot of diagnostics taken
         # immediately before a write.  ``get_diagnostics_sync`` filters
@@ -183,6 +184,9 @@ class LSPService:
         # introduced by the current edit.
         self._delta_baseline: Dict[str, List[Dict[str, Any]]] = {}
 
+        if self._enabled and self._idle_timeout > 0:
+            self._loop.run(self._start_idle_reaper(), timeout=2.0)
+
     @classmethod
     def create_from_config(cls) -> Optional["LSPService"]:
         """Build a service from ``hermes_cli.config`` settings.
@@ -539,6 +543,7 @@ class LSPService:
         with self._state_lock:
             client = self._clients.get(key)
             if client is not None and client.is_running:
+                self._last_used[key] = time.time()
                 eventlog.log_active(srv.server_id, per_server_root)
                 return client
             spawning = self._spawning.get(key)
@@ -597,7 +602,35 @@ class LSPService:
             with self._state_lock:
                 self._spawning.pop(key, None)
 
+    async def _start_idle_reaper(self) -> None:
+        self._idle_reaper_task = asyncio.create_task(self._idle_reaper_loop())
+
+    async def _idle_reaper_loop(self) -> None:
+        interval = min(60.0, self._idle_timeout)
+        while True:
+            await asyncio.sleep(interval)
+            cutoff = time.time() - self._idle_timeout
+            with self._state_lock:
+                idle_keys = [
+                    key
+                    for key in self._clients
+                    if self._last_used.get(key, 0) < cutoff
+                ]
+                clients = [self._clients.pop(key) for key in idle_keys]
+                for key in idle_keys:
+                    self._last_used.pop(key, None)
+            if clients:
+                await asyncio.gather(
+                    *(client.shutdown() for client in clients),
+                    return_exceptions=True,
+                )
+
     async def _shutdown_async(self) -> None:
+        reaper = self._idle_reaper_task
+        self._idle_reaper_task = None
+        if reaper is not None:
+            reaper.cancel()
+            await asyncio.gather(reaper, return_exceptions=True)
         with self._state_lock:
             clients = list(self._clients.values())
             self._clients.clear()

tests/agent/lsp/test_service.py

@@ -8,6 +8,7 @@ on.
 from __future__ import annotations
 
 import sys
+import time
 from pathlib import Path
 
 import pytest
@@ -174,3 +175,33 @@ def test_service_status_includes_clients(mock_pyright):
         assert any(c["server_id"] == "pyright" for c in info["clients"])
     finally:
         svc.shutdown()
+
+
+def test_service_reaps_client_after_idle_timeout(mock_pyright):
+    repo = mock_pyright
+    f = repo / "x.py"
+    f.write_text("")
+    svc = LSPService(
+        enabled=True,
+        wait_mode="document",
+        wait_timeout=3.0,
+        install_strategy="manual",
+        idle_timeout=0.2,
+    )
+    try:
+        svc.get_diagnostics_sync(str(f))
+        assert svc.get_status()["clients"]
+        client = next(iter(svc._clients.values()))
+        process = client._proc
+        assert process is not None
+
+        deadline = time.monotonic() + 2.0
+        while svc.get_status()["clients"] and time.monotonic() < deadline:
+            time.sleep(0.02)
+        while process.returncode is None and time.monotonic() < deadline:
+            time.sleep(0.02)
+
+        assert svc.get_status()["clients"] == []
+        assert process.returncode is not None
+    finally:
+        svc.shutdown()