diff --git a/tools/_lib/__init__.py b/tools/_lib/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tools/_lib/lsp/__init__.py b/tools/_lib/lsp/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tools/_lib/lsp/adapters/__init__.py b/tools/_lib/lsp/adapters/__init__.py new file mode 100644 index 0000000..bf6d72b --- /dev/null +++ b/tools/_lib/lsp/adapters/__init__.py @@ -0,0 +1,22 @@ +"""Adapter registry: detect language from file path, return appropriate adapter.""" + +import os +from .gopls import Gopls +from .clangd import Clangd +from .intelephense import Intelephense + +_ADAPTERS = [Gopls, Clangd, Intelephense] + +_EXT_MAP: dict[str, type] = {} +for _a in _ADAPTERS: + for _ext in _a.extensions: + _EXT_MAP[_ext] = _a + + +def detect(filepath: str): + """Return an adapter class for the given file, or None.""" + ext = os.path.splitext(filepath)[1] + cls = _EXT_MAP.get(ext) + if cls: + return cls() + return None diff --git a/tools/_lib/lsp/adapters/clangd.py b/tools/_lib/lsp/adapters/clangd.py new file mode 100644 index 0000000..9c7df7d --- /dev/null +++ b/tools/_lib/lsp/adapters/clangd.py @@ -0,0 +1,19 @@ +"""clangd adapter.""" + +from ..workspace import find_root + + +class Clangd: + name = "clangd" + extensions = [".c", ".h", ".cc", ".cpp", ".cxx", ".hpp", ".hxx"] + root_markers = ["compile_commands.json", "compile_flags.txt", ".clangd", "CMakeLists.txt", "Makefile"] + cmd = ["clangd", "--background-index"] + + def initialization_options(self): + return {} + + def workspace_root(self, filepath: str, fallback: str | None = None) -> str: + return find_root(filepath, self.root_markers, fallback) + + def prepare_open(self, filepath: str) -> dict | None: + return None diff --git a/tools/_lib/lsp/adapters/gopls.py b/tools/_lib/lsp/adapters/gopls.py new file mode 100644 index 0000000..c7ed160 --- /dev/null +++ b/tools/_lib/lsp/adapters/gopls.py @@ -0,0 +1,24 @@ +"""gopls adapter.""" + +from ..workspace import find_root + + +class Gopls: + name = "gopls" + extensions = [".go"] + root_markers = ["go.work", "go.mod"] + cmd = ["gopls", "serve"] + + def initialization_options(self): + return { + "usePlaceholders": True, + "staticcheck": True, + "gofumpt": True, + } + + def workspace_root(self, filepath: str, fallback: str | None = None) -> str: + return find_root(filepath, self.root_markers, fallback) + + def prepare_open(self, filepath: str) -> dict | None: + """Return extra didOpen params if needed. gopls needs didOpen before queries.""" + return None # handled generically by the bridge diff --git a/tools/_lib/lsp/adapters/intelephense.py b/tools/_lib/lsp/adapters/intelephense.py new file mode 100644 index 0000000..34b45e2 --- /dev/null +++ b/tools/_lib/lsp/adapters/intelephense.py @@ -0,0 +1,19 @@ +"""intelephense adapter (PHP).""" + +from ..workspace import find_root + + +class Intelephense: + name = "intelephense" + extensions = [".php"] + root_markers = ["composer.json", "composer.lock", ".git"] + cmd = ["intelephense", "--stdio"] + + def initialization_options(self): + return {} + + def workspace_root(self, filepath: str, fallback: str | None = None) -> str: + return find_root(filepath, self.root_markers, fallback) + + def prepare_open(self, filepath: str) -> dict | None: + return None diff --git a/tools/_lib/lsp/bridge.py b/tools/_lib/lsp/bridge.py new file mode 100644 index 0000000..0eae31b --- /dev/null +++ b/tools/_lib/lsp/bridge.py @@ -0,0 +1,408 @@ +"""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}, + "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 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 == "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() diff --git a/tools/_lib/lsp/client.py b/tools/_lib/lsp/client.py new file mode 100644 index 0000000..cedb56a --- /dev/null +++ b/tools/_lib/lsp/client.py @@ -0,0 +1,77 @@ +"""Client library for LSP tool scripts. Connects to the bridge daemon, starting it if needed.""" + +import json +import os +import socket +import subprocess +import sys +import time + +SOCK_PATH = os.path.join(os.environ.get("XDG_RUNTIME_DIR", "/tmp"), "ollie-lsp.sock") +PID_PATH = SOCK_PATH + ".pid" + + +def _bridge_running() -> bool: + if not os.path.exists(PID_PATH): + return False + try: + pid = int(open(PID_PATH).read().strip()) + os.kill(pid, 0) + return True + except (OSError, ValueError): + return False + + +def _start_bridge(): + """Start the bridge daemon as a background process.""" + lib_dir = os.path.dirname(os.path.abspath(__file__)) + bridge_script = os.path.join(lib_dir, "bridge.py") + subprocess.Popen( + [sys.executable, bridge_script], + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + # Wait for socket to appear + for _ in range(50): + if os.path.exists(SOCK_PATH): + return + time.sleep(0.1) + print("error: bridge daemon failed to start", file=sys.stderr) + sys.exit(1) + + +def request(method: str, **kwargs) -> dict: + """Send a request to the bridge and return the response.""" + if not _bridge_running(): + _start_bridge() + + payload = {"method": method, **kwargs} + data = json.dumps(payload).encode() + + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + try: + sock.connect(SOCK_PATH) + sock.sendall(data) + sock.shutdown(socket.SHUT_WR) + chunks = [] + while True: + chunk = sock.recv(4096) + if not chunk: + break + chunks.append(chunk) + response = json.loads(b"".join(chunks)) + except ConnectionRefusedError: + # Stale socket, restart + if os.path.exists(SOCK_PATH): + os.unlink(SOCK_PATH) + _start_bridge() + return request(method, **kwargs) + finally: + sock.close() + + if "error" in response: + print(f"error: {response['error']}", file=sys.stderr) + sys.exit(1) + return response.get("result") diff --git a/tools/_lib/lsp/workspace.py b/tools/_lib/lsp/workspace.py new file mode 100644 index 0000000..8a64f13 --- /dev/null +++ b/tools/_lib/lsp/workspace.py @@ -0,0 +1,20 @@ +"""Workspace root detection: walk up from a file to find language-specific markers.""" + +import os +from pathlib import Path + + +def find_root(filepath: str, markers: list[str], fallback: str | None = None) -> str: + """Walk up from filepath looking for any marker file/dir. Return the directory containing it.""" + p = Path(filepath).resolve() + if p.is_file(): + p = p.parent + while True: + for marker in markers: + if (p / marker).exists(): + return str(p) + parent = p.parent + if parent == p: + break + p = parent + return fallback or os.getcwd() diff --git a/tools/lsp_definition b/tools/lsp_definition new file mode 100755 index 0000000..3004e31 --- /dev/null +++ b/tools/lsp_definition @@ -0,0 +1,33 @@ +#!/usr/bin/env python3 +# description: Go to definition of symbol at position. Returns file:line:col. +# Args: + +import os +import sys +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "_lib", "..")) + +from _lib.lsp.client import request + +if len(sys.argv) < 4: + print("usage: lsp_definition ") + sys.exit(1) + +filepath = os.path.abspath(sys.argv[1]) +line = int(sys.argv[2]) - 1 # convert 1-indexed to 0-indexed +col = int(sys.argv[3]) - 1 + +result = request("definition", file=filepath, line=line, col=col) + +if not result: + print("(no definition found)") + sys.exit(0) + +# Normalize: result can be a Location or list of Locations +locations = result if isinstance(result, list) else [result] +for loc in locations: + uri = loc.get("uri", loc.get("targetUri", "")) + path = uri[7:] if uri.startswith("file://") else uri + pos = loc.get("range", loc.get("targetRange", {})).get("start", {}) + ln = pos.get("line", 0) + 1 + ch = pos.get("character", 0) + 1 + print(f"{path}:{ln}:{ch}") diff --git a/tools/lsp_diagnostics b/tools/lsp_diagnostics new file mode 100755 index 0000000..a4cd158 --- /dev/null +++ b/tools/lsp_diagnostics @@ -0,0 +1,34 @@ +#!/usr/bin/env python3 +# description: Get diagnostics (errors, warnings) for a file. +# Args: + +import os +import sys +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "_lib", "..")) + +from _lib.lsp.client import request + +if len(sys.argv) < 2: + print("usage: lsp_diagnostics ") + sys.exit(1) + +filepath = os.path.abspath(sys.argv[1]) + +result = request("diagnostics", file=filepath) + +if not result: + print("(no diagnostics)") + sys.exit(0) + +SEVERITY = {1: "error", 2: "warning", 3: "info", 4: "hint"} + +for diag in result: + sev = SEVERITY.get(diag.get("severity", 3), "info") + ln = diag.get("range", {}).get("start", {}).get("line", 0) + 1 + col = diag.get("range", {}).get("start", {}).get("character", 0) + 1 + msg = diag.get("message", "") + src = diag.get("source", "") + prefix = f"{src}: " if src else "" + print(f"{ln}:{col} {sev}: {prefix}{msg}") + +print(f"\n({len(result)} diagnostics)") diff --git a/tools/lsp_hover b/tools/lsp_hover new file mode 100755 index 0000000..a96538a --- /dev/null +++ b/tools/lsp_hover @@ -0,0 +1,36 @@ +#!/usr/bin/env python3 +# description: Get type/documentation info for symbol at position. +# Args: + +import os +import sys +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "_lib", "..")) + +from _lib.lsp.client import request + +if len(sys.argv) < 4: + print("usage: lsp_hover ") + sys.exit(1) + +filepath = os.path.abspath(sys.argv[1]) +line = int(sys.argv[2]) - 1 +col = int(sys.argv[3]) - 1 + +result = request("hover", file=filepath, line=line, col=col) + +if not result: + print("(no hover info)") + sys.exit(0) + +contents = result.get("contents", "") +if isinstance(contents, dict): + # MarkedString or MarkupContent + print(contents.get("value", str(contents))) +elif isinstance(contents, list): + for item in contents: + if isinstance(item, dict): + print(item.get("value", str(item))) + else: + print(item) +else: + print(contents) diff --git a/tools/lsp_references b/tools/lsp_references new file mode 100755 index 0000000..b6e22e3 --- /dev/null +++ b/tools/lsp_references @@ -0,0 +1,33 @@ +#!/usr/bin/env python3 +# description: Find all references to symbol at position. Returns file:line:col per reference. +# Args: + +import os +import sys +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "_lib", "..")) + +from _lib.lsp.client import request + +if len(sys.argv) < 4: + print("usage: lsp_references ") + sys.exit(1) + +filepath = os.path.abspath(sys.argv[1]) +line = int(sys.argv[2]) - 1 +col = int(sys.argv[3]) - 1 + +result = request("references", file=filepath, line=line, col=col) + +if not result: + print("(no references found)") + sys.exit(0) + +for loc in result: + uri = loc.get("uri", "") + path = uri[7:] if uri.startswith("file://") else uri + pos = loc.get("range", {}).get("start", {}) + ln = pos.get("line", 0) + 1 + ch = pos.get("character", 0) + 1 + print(f"{path}:{ln}:{ch}") + +print(f"\n({len(result)} references)") diff --git a/tools/lsp_rename b/tools/lsp_rename new file mode 100755 index 0000000..3440717 --- /dev/null +++ b/tools/lsp_rename @@ -0,0 +1,32 @@ +#!/usr/bin/env python3 +# description: Rename symbol at position across the workspace. Applies edits directly to files. +# Args: + +import os +import sys +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "_lib", "..")) + +from _lib.lsp.client import request + +if len(sys.argv) < 5: + print("usage: lsp_rename ") + sys.exit(1) + +filepath = os.path.abspath(sys.argv[1]) +line = int(sys.argv[2]) - 1 +col = int(sys.argv[3]) - 1 +new_name = sys.argv[4] + +result = request("rename", file=filepath, line=line, col=col, new_name=new_name) + +if not result: + print("(rename returned no edits)") + sys.exit(0) + +files_changed = result.get("files_changed", {}) +total = result.get("total_edits", 0) + +for path, count in files_changed.items(): + print(f" {path} ({count} edits)") + +print(f"\nRenamed to '{new_name}': {total} edits across {len(files_changed)} files.") diff --git a/tools/lsp_symbols b/tools/lsp_symbols new file mode 100755 index 0000000..f2d4487 --- /dev/null +++ b/tools/lsp_symbols @@ -0,0 +1,55 @@ +#!/usr/bin/env python3 +# description: List symbols in a file or search workspace symbols. +# Args: [--workspace] + +import os +import sys +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "_lib", "..")) + +from _lib.lsp.client import request + +if len(sys.argv) < 2: + print("usage: lsp_symbols [--workspace]") + sys.exit(1) + +workspace = "--workspace" in sys.argv +arg = sys.argv[1] + +SYMBOL_KINDS = { + 1: "File", 2: "Module", 3: "Namespace", 4: "Package", 5: "Class", + 6: "Method", 7: "Property", 8: "Field", 9: "Constructor", 10: "Enum", + 11: "Interface", 12: "Function", 13: "Variable", 14: "Constant", + 15: "String", 16: "Number", 17: "Boolean", 18: "Array", 19: "Object", + 20: "Key", 21: "Null", 22: "EnumMember", 23: "Struct", 24: "Event", + 25: "Operator", 26: "TypeParameter", +} + +if workspace: + result = request("symbols", file=arg, workspace=True, query=arg) +else: + filepath = os.path.abspath(arg) + result = request("symbols", file=filepath, workspace=False) + +if not result: + print("(no symbols found)") + sys.exit(0) + + +def print_symbols(symbols, indent=0): + for sym in symbols: + kind = SYMBOL_KINDS.get(sym.get("kind", 0), "?") + name = sym.get("name", "") + # DocumentSymbol has range; SymbolInformation has location + if "range" in sym: + ln = sym["range"]["start"]["line"] + 1 + print(f"{' ' * indent}{kind:12} {name} (line {ln})") + if "children" in sym: + print_symbols(sym["children"], indent + 1) + elif "location" in sym: + uri = sym["location"].get("uri", "") + path = uri[7:] if uri.startswith("file://") else uri + ln = sym["location"]["range"]["start"]["line"] + 1 + print(f"{kind:12} {name} {path}:{ln}") + + +print_symbols(result)