"""LSP bridge daemon: manages LSP server processes, accepts requests over a Unix socket.""" import asyncio import json import os import signal import sys import time from pathlib import Path from urllib.parse import quote as url_quote # Ensure _lib is importable when running as a standalone daemon _tools_dir = str(Path(__file__).resolve().parent.parent.parent) if _tools_dir not in sys.path: sys.path.insert(0, _tools_dir) SOCK_PATH = os.path.join(os.environ.get("XDG_RUNTIME_DIR", "/tmp"), "ollie-lsp.sock") PID_PATH = SOCK_PATH + ".pid" IDLE_TIMEOUT = 300 # 5 minutes def file_uri(path: str) -> str: return "file://" + os.path.abspath(path) def uri_to_path(uri: str) -> str: if uri.startswith("file://"): return uri[7:] return uri class LspServer: """A running LSP server process.""" def __init__(self, adapter, root: str): self.adapter = adapter self.root = root self.proc: asyncio.subprocess.Process | None = None self.req_id = 0 self.pending: dict[int, asyncio.Future] = {} self.initialized = False self.open_files: set[str] = set() self._read_task: asyncio.Task | None = None self._buf = b"" async def start(self): self.proc = await asyncio.create_subprocess_exec( *self.adapter.cmd, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL, ) self._read_task = asyncio.create_task(self._read_loop()) await self._initialize() async def stop(self): if self.proc and self.proc.returncode is None: await self._request("shutdown", {}) self._notify("exit", None) try: await asyncio.wait_for(self.proc.wait(), timeout=5) except asyncio.TimeoutError: self.proc.kill() if self._read_task: self._read_task.cancel() async def _initialize(self): result = await self._request("initialize", { "processId": os.getpid(), "rootUri": file_uri(self.root), "rootPath": self.root, "capabilities": { "textDocument": { "definition": {"linkSupport": True}, "references": {}, "hover": {"contentFormat": ["plaintext", "markdown"]}, "rename": {"prepareSupport": True}, "documentSymbol": {"hierarchicalDocumentSymbolSupport": True}, "completion": {"completionItem": {"snippetSupport": False}}, "publishDiagnostics": {}, }, "workspace": { "workspaceFolders": True, "symbol": {}, }, }, "initializationOptions": self.adapter.initialization_options(), "workspaceFolders": [{"uri": file_uri(self.root), "name": os.path.basename(self.root)}], }) self._notify("initialized", {}) self.initialized = True return result async def ensure_open(self, filepath: str): if filepath in self.open_files: return try: with open(filepath, "r") as f: text = f.read() except (OSError, UnicodeDecodeError): text = "" ext = os.path.splitext(filepath)[1] lang_map = { ".go": "go", ".c": "c", ".h": "c", ".cc": "cpp", ".cpp": "cpp", ".cxx": "cpp", ".hpp": "cpp", ".hxx": "cpp", ".php": "php", } self._notify("textDocument/didOpen", { "textDocument": { "uri": file_uri(filepath), "languageId": lang_map.get(ext, "plaintext"), "version": 1, "text": text, } }) self.open_files.add(filepath) # Give the server a moment to index the file await asyncio.sleep(0.1) async def definition(self, filepath: str, line: int, col: int): await self.ensure_open(filepath) return await self._request("textDocument/definition", { "textDocument": {"uri": file_uri(filepath)}, "position": {"line": line, "character": col}, }) async def references(self, filepath: str, line: int, col: int): await self.ensure_open(filepath) return await self._request("textDocument/references", { "textDocument": {"uri": file_uri(filepath)}, "position": {"line": line, "character": col}, "context": {"includeDeclaration": True}, }) async def hover(self, filepath: str, line: int, col: int): await self.ensure_open(filepath) return await self._request("textDocument/hover", { "textDocument": {"uri": file_uri(filepath)}, "position": {"line": line, "character": col}, }) async def completion(self, filepath: str, line: int, col: int): await self.ensure_open(filepath) return await self._request("textDocument/completion", { "textDocument": {"uri": file_uri(filepath)}, "position": {"line": line, "character": col}, }) async def rename(self, filepath: str, line: int, col: int, new_name: str): await self.ensure_open(filepath) return await self._request("textDocument/rename", { "textDocument": {"uri": file_uri(filepath)}, "position": {"line": line, "character": col}, "newName": new_name, }) async def document_symbols(self, filepath: str): await self.ensure_open(filepath) return await self._request("textDocument/documentSymbol", { "textDocument": {"uri": file_uri(filepath)}, }) async def workspace_symbols(self, query: str): return await self._request("workspace/symbol", {"query": query}) async def diagnostics(self, filepath: str): await self.ensure_open(filepath) # Most servers push diagnostics via notification; we wait briefly then return cached await asyncio.sleep(1.0) return self._cached_diagnostics.get(file_uri(filepath), []) _cached_diagnostics: dict[str, list] = {} def _notify(self, method: str, params): msg = {"jsonrpc": "2.0", "method": method} if params is not None: msg["params"] = params self._send(msg) async def _request(self, method: str, params: dict, timeout: float = 30.0) -> dict | list | None: self.req_id += 1 rid = self.req_id msg = {"jsonrpc": "2.0", "id": rid, "method": method, "params": params} fut = asyncio.get_event_loop().create_future() self.pending[rid] = fut self._send(msg) try: return await asyncio.wait_for(fut, timeout=timeout) except asyncio.TimeoutError: self.pending.pop(rid, None) raise def _send(self, msg: dict): body = json.dumps(msg).encode() header = f"Content-Length: {len(body)}\r\n\r\n".encode() if self.proc and self.proc.stdin: self.proc.stdin.write(header + body) async def _read_loop(self): try: while self.proc and self.proc.returncode is None: # Read header header_line = await self.proc.stdout.readline() if not header_line: break content_length = 0 while header_line and header_line.strip(): if header_line.startswith(b"Content-Length:"): content_length = int(header_line.split(b":")[1].strip()) header_line = await self.proc.stdout.readline() if content_length == 0: continue body = await self.proc.stdout.readexactly(content_length) msg = json.loads(body) self._handle_message(msg) except (asyncio.CancelledError, asyncio.IncompleteReadError): pass def _handle_message(self, msg: dict): if "id" in msg and "method" not in msg: # Response rid = msg["id"] fut = self.pending.pop(rid, None) if fut and not fut.done(): if "error" in msg: fut.set_exception(RuntimeError(msg["error"].get("message", str(msg["error"])))) else: fut.set_result(msg.get("result")) elif "method" in msg and "id" not in msg: # Notification if msg["method"] == "textDocument/publishDiagnostics": params = msg.get("params", {}) self._cached_diagnostics[params.get("uri", "")] = params.get("diagnostics", []) class Bridge: """The bridge daemon: manages LSP servers, handles tool requests.""" def __init__(self): self.servers: dict[str, LspServer] = {} # key: "adapter_name:root" self.last_activity = time.time() async def handle_request(self, data: dict) -> dict: self.last_activity = time.time() method = data.get("method") filepath = data.get("file") if not filepath: return {"error": "missing 'file' parameter"} filepath = os.path.abspath(filepath) from _lib.lsp.adapters import detect adapter = detect(filepath) if not adapter: return {"error": f"no adapter for file: {filepath}"} fallback = data.get("cwd") root = adapter.workspace_root(filepath, fallback) server = await self._get_server(adapter, root) try: if method == "definition": result = await server.definition(filepath, data["line"], data["col"]) elif method == "references": result = await server.references(filepath, data["line"], data["col"]) elif method == "hover": result = await server.hover(filepath, data["line"], data["col"]) elif method == "completion": result = await server.completion(filepath, data["line"], data["col"]) elif method == "rename": result = await server.rename(filepath, data["line"], data["col"], data["new_name"]) if result: applied = apply_workspace_edit(result) return {"result": applied} return {"result": None} elif method == "symbols": if data.get("workspace"): result = await server.workspace_symbols(data.get("query", "")) else: result = await server.document_symbols(filepath) elif method == "diagnostics": result = await server.diagnostics(filepath) else: return {"error": f"unknown method: {method}"} return {"result": result} except Exception as e: return {"error": str(e)} async def _get_server(self, adapter, root: str) -> LspServer: key = f"{adapter.name}:{root}" if key not in self.servers: server = LspServer(adapter, root) await server.start() self.servers[key] = server return self.servers[key] async def shutdown(self): for server in self.servers.values(): await server.stop() self.servers.clear() def apply_workspace_edit(edit: dict) -> dict: """Apply a WorkspaceEdit to disk. Returns summary of changes.""" changes = {} # Handle documentChanges (versioned) or changes (simple) if "documentChanges" in edit: for doc_change in edit["documentChanges"]: if doc_change.get("kind") in ("create", "rename", "delete"): continue # skip resource operations for now uri = doc_change["textDocument"]["uri"] edits = doc_change.get("edits", []) path = uri_to_path(uri) changes.setdefault(path, []).extend(edits) elif "changes" in edit: for uri, edits in edit["changes"].items(): path = uri_to_path(uri) changes[path] = edits summary = {} for path, edits in changes.items(): try: with open(path, "r") as f: lines = f.readlines() except OSError: continue # Apply edits in reverse order (bottom-up) to preserve positions sorted_edits = sorted(edits, key=lambda e: ( e["range"]["start"]["line"], e["range"]["start"]["character"] ), reverse=True) for e in sorted_edits: start = e["range"]["start"] end = e["range"]["end"] new_text = e["newText"] # Convert line/char positions to string manipulation start_line, start_char = start["line"], start["character"] end_line, end_char = end["line"], end["character"] # Build the new content before = "".join(lines[:start_line]) + lines[start_line][:start_char] if start_line < len(lines) else "".join(lines) after = lines[end_line][end_char:] + "".join(lines[end_line + 1:]) if end_line < len(lines) else "" content = before + new_text + after lines = content.splitlines(True) with open(path, "w") as f: f.writelines(lines) summary[path] = len(sorted_edits) return {"files_changed": summary, "total_edits": sum(summary.values())} async def run_bridge(): bridge = Bridge() async def handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter): try: data = await asyncio.wait_for(reader.read(1024 * 1024), timeout=5.0) if not data: return request = json.loads(data) if request.get("method") == "__shutdown": response = {"result": "ok"} writer.write(json.dumps(response).encode()) await writer.drain() writer.close() raise SystemExit(0) response = await bridge.handle_request(request) writer.write(json.dumps(response).encode()) await writer.drain() except (json.JSONDecodeError, asyncio.TimeoutError) as e: writer.write(json.dumps({"error": str(e)}).encode()) await writer.drain() finally: writer.close() # Clean up stale socket if os.path.exists(SOCK_PATH): os.unlink(SOCK_PATH) server = await asyncio.start_unix_server(handle_client, path=SOCK_PATH) os.chmod(SOCK_PATH, 0o600) # Write PID with open(PID_PATH, "w") as f: f.write(str(os.getpid())) # Idle timeout loop async def idle_watchdog(): while True: await asyncio.sleep(30) if time.time() - bridge.last_activity > IDLE_TIMEOUT: await bridge.shutdown() server.close() return watchdog = asyncio.create_task(idle_watchdog()) try: await server.serve_forever() except (asyncio.CancelledError, SystemExit): pass finally: watchdog.cancel() await bridge.shutdown() if os.path.exists(SOCK_PATH): os.unlink(SOCK_PATH) if os.path.exists(PID_PATH): os.unlink(PID_PATH) def main(): asyncio.run(run_bridge()) if __name__ == "__main__": main()