"""SimVX Language Server -- completes node type names in a Python source file."""
import json
import logging
import sys
from .protocol import decode_header, encode_message, response
log = logging.getLogger(__name__)
_IDENTIFIER_TAIL = set("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_")
[docs]
class SimVXLSPServer:
"""Language server offering one thing: the names of registered node types.
A completion request is answered with the entries of ``Node._registry``
whose name starts with the identifier being typed, each annotated with the
node's declared properties. Anything else -- a member access after a dot, a
position inside a string or a comment -- gets no items, because the server
holds no information that would make an answer better than a guess.
It does not offer diagnostics, member completions resolved against a
node's properties, or input action names. Those need the source parsed and
the enclosing scope resolved, which is scheduled with the editor work
rather than bolted on here.
"""
def __init__(self):
self._running = False
self._initialized = False
self._shutdown_received = False
self._documents: dict[str, str] = {}
self._items: list[dict] = []
self._items_registry_version = -1
# ------------------------------------------------------------------
# Main loop
# ------------------------------------------------------------------
[docs]
def run(self) -> int:
"""Run the server on stdio, reading JSON-RPC messages.
Returns the process exit code: 0 when the client asked for a shutdown
before closing the connection, 1 when it did not.
"""
self._running = True
self._shutdown_received = False
stream = sys.stdin.buffer
while self._running:
msg = self._read_message(stream)
if msg is None:
break
self._handle_message(msg)
return 0 if self._shutdown_received else 1
def _read_message(self, stream) -> dict | None:
"""Read one framed message from *stream*, or ``None`` at end of stream.
The header is read a line at a time up to the blank separator line and
the body is then read by its declared length, so no byte of the stream
is ever scanned twice. Frames with an unusable header or body are
logged and skipped, and reading continues with the next one.
"""
while True:
header = b""
while True:
line = stream.readline()
if not line:
return None
if line in (b"\r\n", b"\n"):
break
header += line
parsed = decode_header(header + b"\r\n\r\n")
if parsed is None:
log.warning("JSON-RPC header without a usable Content-Length")
continue
content_length, _ = parsed
body = b""
while len(body) < content_length:
chunk = stream.read(content_length - len(body))
if not chunk:
return None
body += chunk
try:
msg = json.loads(body)
except json.JSONDecodeError:
log.warning("malformed JSON-RPC body")
continue
if not isinstance(msg, dict):
log.warning("JSON-RPC body is not an object")
continue
return msg
# ------------------------------------------------------------------
# Dispatch
# ------------------------------------------------------------------
_HANDLERS: dict[str, str] = {
"initialize": "_handle_initialize",
"initialized": "_handle_initialized",
"shutdown": "_handle_shutdown",
"exit": "_handle_exit",
"textDocument/didOpen": "_handle_did_open",
"textDocument/didChange": "_handle_did_change",
"textDocument/didClose": "_handle_did_close",
"textDocument/completion": "_handle_completion",
}
def _handle_message(self, msg: dict):
method = msg.get("method", "")
handler_name = self._HANDLERS.get(method)
if handler_name:
getattr(self, handler_name)(msg)
elif "id" in msg and method:
# Unknown request -- respond with method-not-found
self._send(response(msg["id"], error={"code": -32601, "message": f"Method not found: {method}"}))
# ------------------------------------------------------------------
# Lifecycle
# ------------------------------------------------------------------
def _handle_initialize(self, msg: dict):
result = {
"capabilities": {
# No trigger characters: the only completion offered is by
# identifier prefix, so there is no character whose arrival
# tells the server something it could not already answer.
"completionProvider": {},
"textDocumentSync": 1, # Full sync
},
"serverInfo": {"name": "simvx-lsp", "version": "0.1.0"},
}
self._send(response(msg["id"], result=result))
def _handle_initialized(self, _msg: dict):
self._initialized = True
def _handle_shutdown(self, msg: dict):
# The connection stays open after a shutdown: the client is expected to
# follow it with an ``exit`` notification, which ends the loop.
self._send(response(msg["id"], result=None))
self._shutdown_received = True
def _handle_exit(self, _msg: dict):
self._running = False
# ------------------------------------------------------------------
# Document sync
# ------------------------------------------------------------------
def _handle_did_open(self, msg: dict):
params = msg.get("params", {})
doc = params.get("textDocument", {})
uri = doc.get("uri", "")
text = doc.get("text", "")
if uri:
self._documents[uri] = text
def _handle_did_change(self, msg: dict):
params = msg.get("params", {})
uri = params.get("textDocument", {}).get("uri", "")
changes = params.get("contentChanges", [])
if uri and changes:
# Full sync: last change has the full text
self._documents[uri] = changes[-1].get("text", "")
def _handle_did_close(self, msg: dict):
uri = msg.get("params", {}).get("textDocument", {}).get("uri", "")
self._documents.pop(uri, None)
# ------------------------------------------------------------------
# Completions
# ------------------------------------------------------------------
def _handle_completion(self, msg: dict):
params = msg.get("params", {})
uri = params.get("textDocument", {}).get("uri", "")
prefix = self._completion_prefix(self._documents.get(uri, ""), params.get("position", {}))
if prefix is None:
items: list[dict] = []
else:
folded = prefix.casefold()
items = [it for it in self._completion_items() if it["label"].casefold().startswith(folded)]
self._send(response(msg["id"], result={"isIncomplete": False, "items": items}))
@staticmethod
def _completion_prefix(text: str, position: dict) -> str | None:
"""Return the identifier being typed at *position*, or ``None`` to stay silent.
``None`` means the server has nothing useful to say: the word is a
member access, or it sits in a string or a comment. An empty string is
a real answer -- the cursor is at a word boundary, so every node type
is a candidate.
"""
lines = text.split("\n")
row = position.get("line", 0)
if not isinstance(row, int) or not 0 <= row < len(lines):
return None
line = lines[row]
col = position.get("character", 0)
if not isinstance(col, int):
return None
col = max(0, min(col, len(line)))
head = line[:col]
start = len(head)
while start > 0 and head[start - 1] in _IDENTIFIER_TAIL:
start -= 1
# A dot in front makes this an attribute of something the server cannot
# resolve, and quote or hash before the word puts it in text rather than code.
if start > 0 and head[start - 1] == ".":
return None
before = head[:start]
if "#" in before or before.count("'") % 2 or before.count('"') % 2:
return None
return head[start:]
def _completion_items(self) -> list[dict]:
"""Completion items for every registered node type, rebuilt when it changes.
Walking ``dir()`` on every class costs tens of milliseconds, which is
too much to repeat per keystroke. The cache is keyed on the registry's
mutation counter rather than its length: a hot reload rebinds a name to
a new class without changing how many there are, and the new class may
declare different properties.
"""
try:
from ..node import Node
except Exception:
log.debug("Could not load Node registry for completions", exc_info=True)
return []
if Node._registry_version == self._items_registry_version:
return self._items
self._items = self._build_completion_items()
self._items_registry_version = Node._registry_version
return self._items
def _build_completion_items(self) -> list[dict]:
"""Build completion items from the Node registry and Property descriptors."""
items: list[dict] = []
try:
from ..node import Node
for name, cls in sorted(Node._registry.items()):
detail_parts: list[str] = []
# Collect Property descriptors for documentation
props = []
for attr_name in dir(cls):
try:
attr = getattr(cls, attr_name, None)
except Exception:
continue
if attr is not None and type(attr).__name__ == "Property":
props.append(attr_name)
if props:
detail_parts.append(f"Properties: {', '.join(props[:8])}")
if len(props) > 8:
detail_parts.append(f" (+{len(props) - 8} more)")
items.append(
{
"label": name,
"kind": 7, # CompletionItemKind.Class
"detail": "; ".join(detail_parts) if detail_parts else "SimVX node type",
}
)
except Exception:
log.debug("Could not load Node registry for completions", exc_info=True)
return items
# ------------------------------------------------------------------
# Transport
# ------------------------------------------------------------------
def _send(self, msg: dict):
data = encode_message(msg)
sys.stdout.buffer.write(data)
sys.stdout.buffer.flush()