tools: add LSP bridge and tool scripts
Adds a thin LSP bridge daemon (tools/_lib/lsp/) that manages LSP server processes on demand, with adapters for gopls, clangd, and intelephense. Tool scripts: - lsp_definition: go to definition - lsp_references: find all references - lsp_hover: type/doc info - lsp_rename: rename symbol (applies edits directly) - lsp_symbols: document/workspace symbols - lsp_diagnostics: errors and warnings
This commit is contained in:
Levi Neely
parent
cfe367bca1
commit
3fec18b780
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
@ -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()
|
||||
|
|
@ -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")
|
||||
|
|
@ -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()
|
||||
|
|
@ -0,0 +1,33 @@
|
|||
#!/usr/bin/env python3
|
||||
# description: Go to definition of symbol at position. Returns file:line:col.
|
||||
# Args: <file> <line> <col>
|
||||
|
||||
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 <file> <line> <col>")
|
||||
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}")
|
||||
|
|
@ -0,0 +1,34 @@
|
|||
#!/usr/bin/env python3
|
||||
# description: Get diagnostics (errors, warnings) for a file.
|
||||
# Args: <file>
|
||||
|
||||
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 <file>")
|
||||
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)")
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
#!/usr/bin/env python3
|
||||
# description: Get type/documentation info for symbol at position.
|
||||
# Args: <file> <line> <col>
|
||||
|
||||
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 <file> <line> <col>")
|
||||
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)
|
||||
|
|
@ -0,0 +1,33 @@
|
|||
#!/usr/bin/env python3
|
||||
# description: Find all references to symbol at position. Returns file:line:col per reference.
|
||||
# Args: <file> <line> <col>
|
||||
|
||||
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 <file> <line> <col>")
|
||||
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)")
|
||||
|
|
@ -0,0 +1,32 @@
|
|||
#!/usr/bin/env python3
|
||||
# description: Rename symbol at position across the workspace. Applies edits directly to files.
|
||||
# Args: <file> <line> <col> <new_name>
|
||||
|
||||
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 <file> <line> <col> <new_name>")
|
||||
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.")
|
||||
|
|
@ -0,0 +1,55 @@
|
|||
#!/usr/bin/env python3
|
||||
# description: List symbols in a file or search workspace symbols.
|
||||
# Args: <file_or_query> [--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 <file_or_query> [--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)
|
||||
Loading…
Reference in New Issue