feat(lsp): 增加重命名、语义高亮与导入整理

prepareRename/rename 同文件全量出现点;semanticTokens full/range 的 AST+词法+注释 token;organize imports 排序去重。探针新增响应 envelope 校验。
This commit is contained in:
TermiNexus committed 2026-10-26 18:19:05 +08:00
1 parent 7eb20dfae0
commit 08c817f5ec
6 files changed
+1124

No files matched your search

+277
View File
@@ -0,0 +1,277 @@
from stdint import *
import string
import json
import memhub
import ast
import lib.lsp.LspDocuments as LspDocuments
import lib.lsp.LspSource as LspSource
import lib.lsp.LspSymbols as LspSymbols
import lib.lsp.LspWorkspace as LspWorkspace
import lib.lsp.LspNavigation as LspNavigation
import lib.lsp.LspTransport as LspTransport
import lib.lsp.LspRpc as LspRpc
def _IsImportNode(node: ast.AST*) -> int:
if node is None:
return 0
k: int = node.kind()
if k == ast.ASTKind.Import:
return 1
if k == ast.ASTKind.ImportFrom:
return 1
return 0
def _IsDocstring(node: ast.AST*) -> int:
if node is None or node.kind() != ast.ASTKind.Expr:
return 0
ex: ast.Expr* = (ast.Expr*)(node)
if ex.value is None or ex.value.kind() != ast.ASTKind.Constant:
return 0
c: ast.Constant* = (ast.Constant*)(ex.value)
return 1 if c.const_kind == ast.CONST_STR else 0
def _LineText(pool: memhub.MemManager*, doc: LspDocuments.LspDocument*,
line0: int) -> str:
if doc is None or doc.index is None or line0 < 0 or line0 >= doc.index.Count():
return None
sc: int = 0
ec: int = LspSource.LineByteLength(doc, line0)
return LspSource.SliceBytes(pool, doc, line0, sc, line0, ec)
def _TrimLine(pool: memhub.MemManager*, s: str) -> str:
if s is None:
return None
n: sizet = string.strlen(s)
start: sizet = 0
while start < n:
ch: char = s[start]
if ch == ' ' or ch == '\t' or ch == '\r' or ch == '\n':
start += 1
continue
break
end: sizet = n
while end > start:
ch: char = s[end - 1]
if ch == ' ' or ch == '\t' or ch == '\r' or ch == '\n':
end -= 1
continue
break
if end <= start:
return ""
return LspSymbols._CopyStr(pool, s + start, int(end - start))
def _HasUnbalancedParen(s: str) -> int:
if s is None:
return 0
depth: int = 0
i: sizet = 0
while s[i] != 0:
if s[i] == '(':
depth += 1
elif s[i] == ')':
if depth > 0:
depth -= 1
elif s[i] == '\\':
return 1
i += 1
return 1 if depth != 0 else 0
def _Lower(ch: char) -> char:
if ch >= 'A' and ch <= 'Z':
return char(int(ch) + 32)
return ch
def _CompareImport(a: str, b: str) -> int:
if a is None:
return 1 if b is not None else 0
if b is None:
return -1
n: sizet = string.strlen(a)
m: sizet = string.strlen(b)
i: sizet = 0
lim: sizet = n
if m < lim:
lim = m
while i < lim:
ca: char = _Lower(a[i])
cb: char = _Lower(b[i])
if ca < cb:
return -1
if ca > cb:
return 1
i += 1
if n < m:
return -1
if n > m:
return 1
return string.strcmp(a, b)
def _Contains(list_in: list[str]*, s: str) -> int:
if list_in is None or s is None:
return 0
i: int = 0
while i < int(len(list_in)):
if string.strcmp(list_in.get(sizet(i)), s) == 0:
return 1
i += 1
return 0
def _InsertSorted(pool: memhub.MemManager*, out: list[str]*,
s: str, cap: int) -> list[str]*:
new_list: list[str]* = list[str](pool, cap)
if new_list is None:
return out
n: int = int(len(out))
pos: int = 0
while pos < n:
if _CompareImport(s, out.get(sizet(pos))) < 0:
break
pos += 1
i: int = 0
inserted: int = 0
while i < n:
if i == pos:
new_list.append(s)
inserted = 1
new_list.append(out.get(sizet(i)))
i += 1
if inserted == 0:
new_list.append(s)
return new_list
def _Organize(pool: memhub.MemManager*, a: LspSymbols.LspAnalysis*,
uri: str) -> json.JsonValue*:
if a is None or a.tree is None or a.tree.children is None:
return None
body: list[ast.AST*]* = a.tree.children
n: int = int(len(body))
idx: int = 0
if n > 0 and _IsDocstring(body.get(0)) != 0:
idx = 1
first_line: int = -1
last_line: int = -1
originals: list[str]* = list[str](pool, 8)
while idx < n:
st: ast.AST* = body.get(sizet(idx))
if _IsImportNode(st) == 0:
break
if st.lineno <= 0:
return None
line0: int = st.lineno - 1
txt: str = _TrimLine(pool, _LineText(pool, a.doc, line0))
if txt is None or txt[0] == 0 or _HasUnbalancedParen(txt) != 0:
return None
if first_line < 0:
first_line = line0
last_line = line0
if _Contains(originals, txt) == 0:
originals.append(txt)
idx += 1
if first_line < 0:
return None
# Only organize a contiguous import block: reject imports later in the file.
while idx < n:
if _IsImportNode(body.get(sizet(idx))) != 0:
return None
idx += 1
sorted_imports: list[str]* = list[str](pool, 8)
i: int = 0
while i < int(len(originals)):
sorted_imports = _InsertSorted(pool, sorted_imports, originals.get(sizet(i)), 64)
i += 1
changed: int = 0
if int(len(sorted_imports)) != int(len(originals)):
changed = 1
else:
i = 0
while i < int(len(sorted_imports)):
if string.strcmp(sorted_imports.get(sizet(i)), originals.get(sizet(i))) != 0:
changed = 1
break
i += 1
if changed == 0:
return None
new_text: str = None
i = 0
while i < int(len(sorted_imports)):
if new_text is None:
new_text = LspSymbols._CopyStr(pool, sorted_imports.get(sizet(i)), 1024)
else:
new_text = _Join(pool, new_text, "\n",
sorted_imports.get(sizet(i)))
i += 1
if new_text is None:
return None
rng: LspSource.LspRange* = LspSource.SpanFromBytes(
pool, a.doc, first_line, 0, last_line, LspSource.LineByteLength(a.doc, last_line))
edit: json.JsonValue* = json.object(pool)
if edit is None:
return None
json.object_set(pool, edit, "range", LspSource.RangeToJson(pool, rng))
json.object_set(pool, edit, "newText", json.string_val(pool, new_text))
edits: json.JsonValue* = json.array(pool)
if edits is None:
return None
json.array_append(pool, edits, edit)
changes: json.JsonValue* = json.object(pool)
if changes is None:
return None
json.object_set(pool, changes, uri, edits)
we: json.JsonValue* = json.object(pool)
if we is None:
return None
json.object_set(pool, we, "changes", changes)
action: json.JsonValue* = json.object(pool)
if action is None:
return None
json.object_set(pool, action, "title", json.string_val(pool, "Organize imports"))
json.object_set(pool, action, "kind", json.string_val(pool, "source.organizeImports"))
json.object_set(pool, action, "edit", we)
return action
def _Join(pool: memhub.MemManager*, a: str, sep: str, b: str) -> str:
if a is None:
return b
if b is None:
return a
n1: sizet = string.strlen(a)
n2: sizet = string.strlen(sep)
n3: sizet = string.strlen(b)
buf: bytes = pool.alloc(n1 + n2 + n3 + 1)
if buf is None:
return None
string.memcpy(buf, a, n1)
string.memcpy(buf + n1, sep, n2)
string.memcpy(buf + n1 + n2, b, n3)
buf[n1 + n2 + n3] = 0
return str(buf)
def CodeAction(pool: memhub.MemManager*, tr: LspTransport.LspTransport*,
docs: LspDocuments.LspDocuments*, ws: LspWorkspace.LspWorkspace*,
id_node: json.JsonValue*, params: json.JsonValue*) -> int:
arr: json.JsonValue* = json.array(pool)
if arr is None:
return -1
if params is None:
return LspRpc.WriteResponse(pool, tr, id_node, arr)
doc: LspDocuments.LspDocument* = LspNavigation._Doc(pool, docs, params)
if doc is None:
return LspRpc.WriteResponse(pool, tr, id_node, arr)
uri: str = LspNavigation._Uri(params)
a: LspSymbols.LspAnalysis* = LspNavigation._Analyze(pool, docs, doc, uri)
action: json.JsonValue* = _Organize(pool, a, uri)
if action is not None:
json.array_append(pool, arr, action)
return LspRpc.WriteResponse(pool, tr, id_node, arr)
+230
View File
@@ -0,0 +1,230 @@
from stdint import *
import string
import json
import memhub
import ast
import lib.lsp.LspDocuments as LspDocuments
import lib.lsp.LspSource as LspSource
import lib.lsp.LspSymbols as LspSymbols
import lib.lsp.LspWorkspace as LspWorkspace
import lib.lsp.LspNavigation as LspNavigation
import lib.lsp.LspTransport as LspTransport
import lib.lsp.LspRpc as LspRpc
def _IsReserved(name: str) -> int:
if name is None:
return 1
if string.strcmp(name, "False") == 0: return 1
if string.strcmp(name, "None") == 0: return 1
if string.strcmp(name, "True") == 0: return 1
if string.strcmp(name, "and") == 0: return 1
if string.strcmp(name, "as") == 0: return 1
if string.strcmp(name, "assert") == 0: return 1
if string.strcmp(name, "async") == 0: return 1
if string.strcmp(name, "await") == 0: return 1
if string.strcmp(name, "break") == 0: return 1
if string.strcmp(name, "case") == 0: return 1
if string.strcmp(name, "class") == 0: return 1
if string.strcmp(name, "continue") == 0: return 1
if string.strcmp(name, "def") == 0: return 1
if string.strcmp(name, "del") == 0: return 1
if string.strcmp(name, "elif") == 0: return 1
if string.strcmp(name, "else") == 0: return 1
if string.strcmp(name, "except") == 0: return 1
if string.strcmp(name, "finally") == 0: return 1
if string.strcmp(name, "for") == 0: return 1
if string.strcmp(name, "from") == 0: return 1
if string.strcmp(name, "global") == 0: return 1
if string.strcmp(name, "if") == 0: return 1
if string.strcmp(name, "import") == 0: return 1
if string.strcmp(name, "in") == 0: return 1
if string.strcmp(name, "is") == 0: return 1
if string.strcmp(name, "lambda") == 0: return 1
if string.strcmp(name, "match") == 0: return 1
if string.strcmp(name, "nonlocal") == 0: return 1
if string.strcmp(name, "not") == 0: return 1
if string.strcmp(name, "or") == 0: return 1
if string.strcmp(name, "pass") == 0: return 1
if string.strcmp(name, "raise") == 0: return 1
if string.strcmp(name, "return") == 0: return 1
if string.strcmp(name, "try") == 0: return 1
if string.strcmp(name, "while") == 0: return 1
if string.strcmp(name, "with") == 0: return 1
if string.strcmp(name, "yield") == 0: return 1
# Viper modifiers/soft keywords that cannot be identifiers in declaration position.
if string.strcmp(name, "extern") == 0: return 1
if string.strcmp(name, "export") == 0: return 1
if string.strcmp(name, "static") == 0: return 1
if string.strcmp(name, "virtual") == 0: return 1
if string.strcmp(name, "novirtual") == 0: return 1
if string.strcmp(name, "inline") == 0: return 1
if string.strcmp(name, "define") == 0: return 1
if string.strcmp(name, "type") == 0: return 1
return 0
def _ValidName(name: str) -> int:
if name is None or name[0] == 0:
return 0
n: sizet = string.strlen(name)
if n > 128:
return 0
i: sizet = 0
while i < n:
ch: char = name[i]
if i == 0:
if LspSource._IsIdentStart(ch) == 0:
return 0
else:
if LspSource._IsIdentChar(ch) == 0:
return 0
i += 1
return 0 if _IsReserved(name) != 0 else 1
def _OccRange(pool: memhub.MemManager*,
o: LspSymbols.LspOccurrence*) -> LspSource.LspRange*:
if o is None:
return None
return LspSource.NewRange(pool, o.line, o.sc, o.line, o.ec)
def _TextOfRange(pool: memhub.MemManager*, a: LspSymbols.LspAnalysis*,
r: LspSource.LspRange*) -> str:
if r is None:
return None
sc: int = LspSource.Utf16ColToByte(a.doc, r.sl, r.sc)
ec: int = LspSource.Utf16ColToByte(a.doc, r.el, r.ec)
return LspSource.SliceBytes(pool, a.doc, r.sl, sc, r.el, ec)
def _SameRange(a: LspSource.LspRange*, b: LspSource.LspRange*) -> int:
if a is None or b is None:
return 0
if a.sl != b.sl or a.sc != b.sc or a.el != b.el or a.ec != b.ec:
return 0
return 1
def _FindTarget(a: LspSymbols.LspAnalysis*, line0: int,
ch: int) -> LspSymbols.LspSymbol*:
occ: LspSymbols.LspOccurrence* = LspNavigation._OccAt(a, line0, ch)
if occ is not None and occ.sym is not None:
return occ.sym
return LspNavigation._SymAt(a, line0, ch)
def _TokenRange(pool: memhub.MemManager*, a: LspSymbols.LspAnalysis*,
line0: int, ch: int) -> LspSource.LspRange*:
occ: LspSymbols.LspOccurrence* = LspNavigation._OccAt(a, line0, ch)
if occ is not None:
return _OccRange(pool, occ)
s: LspSymbols.LspSymbol* = LspNavigation._SymAt(a, line0, ch)
if s is not None and s.sel is not None:
return s.sel
return None
def Prepare(pool: memhub.MemManager*, tr: LspTransport.LspTransport*,
docs: LspDocuments.LspDocuments*, ws: LspWorkspace.LspWorkspace*,
id_node: json.JsonValue*, params: json.JsonValue*) -> int:
if params is None:
return LspRpc.WriteResponse(pool, tr, id_node, json.null(pool))
doc: LspDocuments.LspDocument* = LspNavigation._Doc(pool, docs, params)
pos: json.JsonValue* = LspNavigation._Position(params)
if doc is None or pos is None:
return LspRpc.WriteResponse(pool, tr, id_node, json.null(pool))
line0: int = LspRpc.GetIntField(pos, "line", 0)
ch: int = LspRpc.GetIntField(pos, "character", 0)
a: LspSymbols.LspAnalysis* = LspNavigation._Analyze(
pool, docs, doc, LspNavigation._Uri(params))
sym: LspSymbols.LspSymbol* = _FindTarget(a, line0, ch)
if sym is None or sym.name is None:
return LspRpc.WriteResponse(pool, tr, id_node, json.null(pool))
if sym.file is None or a.file is None or string.strcmp(sym.file, a.file) != 0:
return LspRpc.WriteResponse(pool, tr, id_node, json.null(pool))
token: LspSource.LspRange* = _TokenRange(pool, a, line0, ch)
if token is None:
return LspRpc.WriteResponse(pool, tr, id_node, json.null(pool))
result: json.JsonValue* = json.object(pool)
if result is None:
return -1
json.object_set(pool, result, "range", LspSource.RangeToJson(pool, token))
json.object_set(pool, result, "placeholder", json.string_val(pool, sym.name))
return LspRpc.WriteResponse(pool, tr, id_node, result)
def Rename(pool: memhub.MemManager*, tr: LspTransport.LspTransport*,
docs: LspDocuments.LspDocuments*, ws: LspWorkspace.LspWorkspace*,
id_node: json.JsonValue*, params: json.JsonValue*) -> int:
if params is None:
return LspRpc.WriteResponse(pool, tr, id_node, json.null(pool))
new_name: str = LspRpc.GetStringField(params, "newName")
if _ValidName(new_name) == 0:
return LspRpc.WriteResponse(pool, tr, id_node, json.null(pool))
doc: LspDocuments.LspDocument* = LspNavigation._Doc(pool, docs, params)
pos: json.JsonValue* = LspNavigation._Position(params)
if doc is None or pos is None:
return LspRpc.WriteResponse(pool, tr, id_node, json.null(pool))
line0: int = LspRpc.GetIntField(pos, "line", 0)
ch: int = LspRpc.GetIntField(pos, "character", 0)
uri: str = LspNavigation._Uri(params)
a: LspSymbols.LspAnalysis* = LspNavigation._Analyze(pool, docs, doc, uri)
sym: LspSymbols.LspSymbol* = _FindTarget(a, line0, ch)
if sym is None or a is None:
return LspRpc.WriteResponse(pool, tr, id_node, json.null(pool))
if sym.file is None or a.file is None or string.strcmp(sym.file, a.file) != 0:
return LspRpc.WriteResponse(pool, tr, id_node, json.null(pool))
edits: json.JsonValue* = json.array(pool)
if edits is None:
return -1
seen: list[LspSource.LspRange*]* = list[LspSource.LspRange*](pool, 16)
# The declaration itself (parameters, functions, classes, fields, locals).
if sym.sel is not None:
txt: str = _TextOfRange(pool, a, sym.sel)
if txt is not None and string.strcmp(txt, sym.name) == 0:
seen.append(sym.sel)
_AppendEdit(pool, edits, sym.sel, new_name)
i: int = 0
while i < int(len(a.occs)):
o: LspSymbols.LspOccurrence* = a.occs.get(sizet(i))
i += 1
if o is None or o.sym != sym:
continue
r: LspSource.LspRange* = _OccRange(pool, o)
if r is None:
continue
dup: int = 0
j: int = 0
while j < int(len(seen)):
if _SameRange(seen.get(sizet(j)), r) != 0:
dup = 1
break
j += 1
if dup != 0:
continue
seen.append(r)
_AppendEdit(pool, edits, r, new_name)
if int(len(edits)) == 0:
return LspRpc.WriteResponse(pool, tr, id_node, json.null(pool))
changes: json.JsonValue* = json.object(pool)
if changes is None:
return -1
json.object_set(pool, changes, uri, edits)
result: json.JsonValue* = json.object(pool)
if result is None:
return -1
json.object_set(pool, result, "changes", changes)
return LspRpc.WriteResponse(pool, tr, id_node, result)
def _AppendEdit(pool: memhub.MemManager*, arr: json.JsonValue*,
r: LspSource.LspRange*, text: str):
edit: json.JsonValue* = json.object(pool)
if edit is None:
return
json.object_set(pool, edit, "range", LspSource.RangeToJson(pool, r))
json.object_set(pool, edit, "newText", json.string_val(pool, text))
json.array_append(pool, arr, edit)
+354
View File
@@ -0,0 +1,354 @@
from stdint import *
import string
import json
import memhub
import ast
import lib.lsp.LspDocuments as LspDocuments
import lib.lsp.LspSource as LspSource
import lib.lsp.LspSymbols as LspSymbols
import lib.lsp.LspWorkspace as LspWorkspace
import lib.lsp.LspNavigation as LspNavigation
import lib.lsp.LspTransport as LspTransport
import lib.lsp.LspRpc as LspRpc
define LSP_MAX_SEMANTIC_TOKENS = 10000
# Indices must match the legend built by Provider().
define ST_NAMESPACE = 0
define ST_CLASS = 1
define ST_FUNCTION = 2
define ST_METHOD = 3
define ST_PARAMETER = 4
define ST_VARIABLE = 5
define ST_PROPERTY = 6
define ST_KEYWORD = 7
define ST_COMMENT = 8
define ST_STRING = 9
define ST_NUMBER = 10
define ST_OPERATOR = 11
define ST_DECORATOR = 12
class LspSemToken:
sl: int
sc: int
el: int
ec: int
typ: int
def __new__(self, pool: memhub.MemManager*) -> rawptr:
p: LspSemToken* = (LspSemToken*)(pool.alloc(LspSemToken.__sizeof__()))
if p:
string.memset(p, 0, LspSemToken.__sizeof__())
return rawptr(p)
def _NewTok(pool: memhub.MemManager*, sl: int, sc: int, el: int, ec: int,
typ: int) -> LspSemToken*:
t: LspSemToken* = LspSemToken(pool)
if t is None:
return None
t.sl = sl
t.sc = sc
t.el = el
t.ec = ec
t.typ = typ
return t
def Provider(pool: memhub.MemManager*) -> json.JsonValue*:
provider: json.JsonValue* = json.object(pool)
legend: json.JsonValue* = json.object(pool)
types: json.JsonValue* = json.array(pool)
mods: json.JsonValue* = json.array(pool)
if provider is None or legend is None or types is None or mods is None:
return None
json.array_append(pool, types, json.string_val(pool, "namespace"))
json.array_append(pool, types, json.string_val(pool, "class"))
json.array_append(pool, types, json.string_val(pool, "function"))
json.array_append(pool, types, json.string_val(pool, "method"))
json.array_append(pool, types, json.string_val(pool, "parameter"))
json.array_append(pool, types, json.string_val(pool, "variable"))
json.array_append(pool, types, json.string_val(pool, "property"))
json.array_append(pool, types, json.string_val(pool, "keyword"))
json.array_append(pool, types, json.string_val(pool, "comment"))
json.array_append(pool, types, json.string_val(pool, "string"))
json.array_append(pool, types, json.string_val(pool, "number"))
json.array_append(pool, types, json.string_val(pool, "operator"))
json.array_append(pool, types, json.string_val(pool, "decorator"))
json.array_append(pool, mods, json.string_val(pool, "declaration"))
json.object_set(pool, legend, "tokenTypes", types)
json.object_set(pool, legend, "tokenModifiers", mods)
json.object_set(pool, provider, "legend", legend)
json.object_set(pool, provider, "full", json.bool_val(pool, True))
json.object_set(pool, provider, "range", json.bool_val(pool, True))
return provider
def _TypeFor(s: LspSymbols.LspSymbol*) -> int:
if s is None:
return ST_VARIABLE
if s.kind == LspSymbols.LSP_KIND_MODULE:
return ST_NAMESPACE
if s.kind == LspSymbols.LSP_KIND_CLASS:
return ST_CLASS
if s.kind == LspSymbols.LSP_KIND_FUNCTION:
return ST_FUNCTION
if s.kind == LspSymbols.LSP_KIND_METHOD:
return ST_METHOD
if s.kind == LspSymbols.LSP_KIND_FIELD:
return ST_PROPERTY
if s.decl is not None and s.decl.kind() == ast.ASTKind.Arg:
return ST_PARAMETER
return ST_VARIABLE
def _TokenForSymbol(pool: memhub.MemManager*,
s: LspSymbols.LspSymbol*) -> LspSemToken*:
if s is None or s.sel is None:
return None
return _NewTok(pool, s.sel.sl, s.sel.sc, s.sel.el, s.sel.ec, _TypeFor(s))
def _TokenForOcc(pool: memhub.MemManager*,
o: LspSymbols.LspOccurrence*) -> LspSemToken*:
if o is None or o.sym is None:
return None
return _NewTok(pool, o.line, o.sc, o.line, o.ec, _TypeFor(o.sym))
def _Before(a: LspSemToken*, b: LspSemToken*) -> int:
if a.sl < b.sl:
return 1
if a.sl > b.sl:
return 0
return 1 if a.sc <= b.sc else 0
def _Overlaps(last: LspSemToken*, t: LspSemToken*) -> int:
if last is None:
return 0
if t.sl < last.el:
return 1
if t.sl == last.el and t.sc < last.ec:
return 1
return 0
def _AstStream(pool: memhub.MemManager*,
a: LspSymbols.LspAnalysis*) -> list[LspSemToken*]*:
out: list[LspSemToken*]* = list[LspSemToken*](pool, 64)
syms: list[LspSymbols.LspSymbol*]* = a.symbols
occs: list[LspSymbols.LspOccurrence*]* = a.occs
si: int = 0
oi: int = 0
while si < int(len(syms)) or oi < int(len(occs)):
sym_t: LspSemToken* = None
if si < int(len(syms)):
sym_t = _TokenForSymbol(pool, syms.get(sizet(si)))
occ_t: LspSemToken* = None
if oi < int(len(occs)):
occ_t = _TokenForOcc(pool, occs.get(sizet(oi)))
if sym_t is None and occ_t is None:
if sym_t is None:
si += 1
if occ_t is None:
oi += 1
continue
if sym_t is None:
out.append(occ_t)
oi += 1
continue
if occ_t is None:
out.append(sym_t)
si += 1
continue
if _Before(sym_t, occ_t) != 0:
out.append(sym_t)
si += 1
else:
out.append(occ_t)
oi += 1
return out
def _ScanComments(pool: memhub.MemManager*, doc: LspDocuments.LspDocument*,
from_off: int, to_off: int, out: list[LspSemToken*]*):
off: int = from_off
total: int = int(doc.index.Length())
if to_off > total:
to_off = total
while off < to_off:
ch: char = doc.text[off]
if ch == '\'' or ch == '"':
off = int(LspSource._SkipQuoted(doc, sizet(off), sizet(to_off)))
continue
if ch == '#':
start: int = off
while off < to_off and doc.text[off] != '\n':
off += 1
l: int = 0
c: int = 0
doc.index.OffsetToPosition(sizet(start), &l, &c)
ec: int = LspSource.ByteColToUtf16(doc, l, int(sizet(off) - doc.index.LineStart(l)))
if ec > c:
t: LspSemToken* = _NewTok(pool, l, c, l, ec, ST_COMMENT)
if t is not None:
out.append(t)
continue
off += 1
def _LexStream(pool: memhub.MemManager*,
doc: LspDocuments.LspDocument*) -> list[LspSemToken*]*:
out: list[LspSemToken*]* = list[LspSemToken*](pool, 128)
lx: ast.Lexer* = ast.new_lexer(pool)
if lx is None:
return out
ast._lexer_init(lx, doc.text, pool)
tok: ast.Token* = ast.tokenize(lx)
total: int = int(doc.index.Length())
last_off: int = 0
while tok is not None:
if tok.lineno <= 0:
tok = tok.next
continue
line0: int = tok.lineno - 1
if line0 >= doc.index.Count():
tok = tok.next
continue
tok_off: int = int(doc.index.LineStart(line0)) + tok.col_offset
if tok_off > last_off:
_ScanComments(pool, doc, last_off, tok_off, out)
typ: int = -1
if tok.type == ast.TokenType.Name:
if tok.kw_subtype != ast.Keyword.NotKeyword:
typ = ST_KEYWORD
elif tok.type == ast.TokenType.String:
typ = ST_STRING
elif tok.type == ast.TokenType.FStringStart:
typ = ST_STRING
elif tok.type == ast.TokenType.FStringMiddle:
typ = ST_STRING
elif tok.type == ast.TokenType.FStringEnd:
typ = ST_STRING
elif tok.type == ast.TokenType.Number:
typ = ST_NUMBER
end_line0: int = line0
end_col: int = tok.col_offset
if tok.end_lineno > tok.lineno and tok.end_col_offset > 0:
end_line0 = tok.end_lineno - 1
end_col = tok.end_col_offset
elif tok.end_lineno == tok.lineno and tok.end_col_offset > tok.col_offset:
end_col = tok.end_col_offset
elif tok.str_val is not None:
end_col = tok.col_offset + int(string.strlen(tok.str_val))
if end_col < tok.col_offset:
end_col = tok.col_offset
if typ >= 0 and end_col > tok.col_offset:
sc: int = LspSource.ByteColToUtf16(doc, line0, tok.col_offset)
ec: int = LspSource.ByteColToUtf16(doc, end_line0, end_col)
if ec > sc:
t: LspSemToken* = _NewTok(pool, line0, sc, end_line0, ec, typ)
if t is not None:
out.append(t)
new_off: int = int(doc.index.LineStart(end_line0)) + end_col
if new_off > last_off:
last_off = new_off
tok = tok.next
if total > last_off:
_ScanComments(pool, doc, last_off, total, out)
return out
def _InRange(t: LspSemToken*, start_line: int, end_line: int) -> int:
if t is None:
return 0
if t.sl < start_line:
return 0
if t.sl > end_line:
return 0
return 1
def _Encode(pool: memhub.MemManager*, out: json.JsonValue*,
ast_tokens: list[LspSemToken*]*, lex_tokens: list[LspSemToken*]*,
has_range: int, start_line: int, end_line: int):
prev_line: int = 0
prev_char: int = 0
count: int = 0
last: LspSemToken* = None
ai: int = 0
li: int = 0
an: int = int(len(ast_tokens))
ln: int = int(len(lex_tokens))
while (ai < an or li < ln) and count < LSP_MAX_SEMANTIC_TOKENS:
at: LspSemToken* = ast_tokens.get(sizet(ai)) if ai < an else None
lt: LspSemToken* = lex_tokens.get(sizet(li)) if li < ln else None
t: LspSemToken* = None
if at is None:
t = lt
li += 1
elif lt is None:
t = at
ai += 1
elif _Before(at, lt) != 0:
t = at
ai += 1
else:
t = lt
li += 1
if t is None:
continue
if has_range != 0 and _InRange(t, start_line, end_line) == 0:
continue
if _Overlaps(last, t) != 0:
continue
dl: int = t.sl - prev_line
dc: int = t.sc - prev_char if dl == 0 else t.sc
length: int = t.ec - t.sc
if length <= 0:
continue
json.array_append(pool, out, json.int_val(pool, int64(dl)))
json.array_append(pool, out, json.int_val(pool, int64(dc)))
json.array_append(pool, out, json.int_val(pool, int64(length)))
json.array_append(pool, out, json.int_val(pool, int64(t.typ)))
json.array_append(pool, out, json.int_val(pool, 0))
prev_line = t.sl
prev_char = t.sc
last = t
count += 1
def Tokens(pool: memhub.MemManager*, tr: LspTransport.LspTransport*,
docs: LspDocuments.LspDocuments*, ws: LspWorkspace.LspWorkspace*,
id_node: json.JsonValue*, params: json.JsonValue*,
range_only: int) -> int:
result: json.JsonValue* = json.object(pool)
data: json.JsonValue* = json.array(pool)
if result is None or data is None:
return -1
doc: LspDocuments.LspDocument* = LspNavigation._Doc(pool, docs, params)
if doc is None:
json.object_set(pool, result, "data", data)
return LspRpc.WriteResponse(pool, tr, id_node, result)
a: LspSymbols.LspAnalysis* = LspNavigation._Analyze(
pool, docs, doc, LspNavigation._Uri(params))
ast_tokens: list[LspSemToken*]* = _AstStream(pool, a)
lex_tokens: list[LspSemToken*]* = _LexStream(pool, doc)
has_range: int = 0
start_line: int = 0
end_line: int = 0
if range_only != 0 and params is not None:
rng: json.JsonValue* = params["range"]
if rng is not None:
start: json.JsonValue* = rng["start"]
end: json.JsonValue* = rng["end"]
if start is not None and end is not None:
has_range = 1
start_line = LspRpc.GetIntField(start, "line", 0)
end_line = LspRpc.GetIntField(end, "line", 0)
_Encode(pool, data, ast_tokens, lex_tokens, has_range, start_line, end_line)
json.object_set(pool, result, "data", data)
return LspRpc.WriteResponse(pool, tr, id_node, result)
+22
View File
@@ -16,6 +16,9 @@ import lib.lsp.LspIndex as LspIndex
import lib.lsp.LspNavigation as LspNavigation
import lib.lsp.LspCompletion as LspCompletion
import lib.lsp.LspSignature as LspSignature
import lib.lsp.LspRename as LspRename
import lib.lsp.LspSemantic as LspSemantic
import lib.lsp.LspCodeAction as LspCodeAction
define LSP_SERVER_NAME = "viperc-lsp"
@@ -65,6 +68,15 @@ def _BuildInitializeResult(pool: memhub.MemManager*) -> json.JsonValue*:
json.array_append(pool, sig_trig, json.string_val(pool, ","))
json.object_set(pool, sig, "triggerCharacters", sig_trig)
json.object_set(pool, caps, "signatureHelpProvider", sig)
rename: json.JsonValue* = json.object(pool)
if rename is not None:
json.object_set(pool, rename, "prepareProvider", json.bool_val(pool, True))
json.object_set(pool, caps, "renameProvider", rename)
sem: json.JsonValue* = LspSemantic.Provider(pool)
if sem is not None:
json.object_set(pool, caps, "semanticTokensProvider", sem)
json.object_set(pool, caps, "codeActionProvider", json.bool_val(pool, True))
json.object_set(pool, result, "capabilities", caps)
json.object_set(pool, info, "name", name)
json.object_set(pool, info, "version", ver)
@@ -412,6 +424,16 @@ def Run(pool: memhub.MemBuddy*, perm_pool: memhub.MemBuddy*) -> int:
rc = LspCompletion.Completion(pool, tr, docs, ws, id_node, params)
elif string.strcmp(method, "textDocument/signatureHelp") == 0:
rc = LspSignature.SignatureHelp(pool, tr, docs, ws, id_node, params)
elif string.strcmp(method, "textDocument/prepareRename") == 0:
rc = LspRename.Prepare(pool, tr, docs, ws, id_node, params)
elif string.strcmp(method, "textDocument/rename") == 0:
rc = LspRename.Rename(pool, tr, docs, ws, id_node, params)
elif string.strcmp(method, "textDocument/semanticTokens/full") == 0:
rc = LspSemantic.Tokens(pool, tr, docs, ws, id_node, params, 0)
elif string.strcmp(method, "textDocument/semanticTokens/range") == 0:
rc = LspSemantic.Tokens(pool, tr, docs, ws, id_node, params, 1)
elif string.strcmp(method, "textDocument/codeAction") == 0:
rc = LspCodeAction.CodeAction(pool, tr, docs, ws, id_node, params)
elif string.strcmp(method, "viper/lspStats") == 0:
rc = _HandleStats(pool, tr, docs, ws, id_node)
elif string.strcmp(method, "viper/lspDocument") == 0:
+24
View File
@@ -0,0 +1,24 @@
from .gate3_stub import StubThing
from .gate3_helper import helper_func
from .gate3_helper import helper_func
import stdlib
VALUE: int = 42
NAME: str = "hello"
class Thing:
field: int
def __init__(self):
self.field = 1
def method(self, arg: int) -> int:
local: int = arg + self.field
return local
def use(x: int) -> int:
# gate5 comment
t: Thing = Thing()
return t.field + x
+217
View File
@@ -106,6 +106,12 @@ class Client:
continue
if response.get("id") != self.next_id:
raise AssertionError(f"id mismatch: {response}")
if response.get("method") is not None:
raise AssertionError(f"response carries a method field: {response}")
if "result" not in response and "error" not in response:
raise AssertionError(f"message is not a response: {response}")
if response.get("jsonrpc") != "2.0":
raise AssertionError(f"response has no jsonrpc 2.0: {response}")
return response
def wait_notification(self, method: str) -> dict[str, Any]:
@@ -742,6 +748,214 @@ def gate4(exe: str) -> int:
return 0
def _apply_edits(text: str, edits: list[dict[str, Any]]) -> str:
lines = text.split("\n")
ordered = sorted(
edits,
key=lambda e: (e["range"]["start"]["line"], e["range"]["start"]["character"]),
reverse=True,
)
for edit in ordered:
start = edit["range"]["start"]
end = edit["range"]["end"]
sl = start["line"]
sc = start["character"]
el = end["line"]
ec = end["character"]
prefix = lines[sl][:sc] if sl < len(lines) else ""
suffix = lines[el][ec:] if el < len(lines) else ""
replacement = prefix + edit["newText"] + suffix
if sl < len(lines):
lines[sl] = replacement
else:
lines.append(replacement)
if el > sl:
del lines[sl + 1 : el + 1]
return "\n".join(lines)
def _decode_tokens(data: list[int]) -> list[dict[str, int]]:
if not isinstance(data, list) or len(data) % 5 != 0:
raise AssertionError(f"semantic token data is malformed: {data}")
tokens = []
line = 0
char = 0
for i in range(0, len(data), 5):
delta_line, delta_char, length, token_type, modifiers = data[i : i + 5]
line += delta_line
char = char + delta_char if delta_line == 0 else delta_char
tokens.append(
{
"line": line,
"character": char,
"length": length,
"type": token_type,
"modifiers": modifiers,
}
)
return tokens
def _find_token(tokens: list[dict[str, int]], line: int, character: int,
token_type: int) -> bool:
return any(
t["line"] == line and t["character"] == character and t["type"] == token_type
for t in tokens
)
def gate5(exe: str) -> int:
root = Path(__file__).resolve().parents[2]
source = root / "Tools" / "lsp" / "fixtures" / "gate5.vp"
uri = source.as_uri()
text = source.read_text(encoding="utf-8")
client = Client(exe)
response = client.request(
"initialize",
{"processId": os.getpid(), "rootUri": root.as_uri(), "capabilities": {}},
)
capabilities = response.get("result", {}).get("capabilities", {})
rename_provider = capabilities.get("renameProvider")
if not isinstance(rename_provider, dict) or rename_provider.get("prepareProvider") is not True:
raise AssertionError(f"missing prepareRename capability: {response}")
sem = capabilities.get("semanticTokensProvider")
if not isinstance(sem, dict) or sem.get("full") is not True or sem.get("range") is not True:
raise AssertionError(f"missing semanticTokens capability: {response}")
legend = sem.get("legend", {})
types = legend.get("tokenTypes", [])
for required in ("class", "function", "parameter", "keyword", "comment", "string", "number"):
if required not in types:
raise AssertionError(f"semantic legend misses {required}: {legend}")
if capabilities.get("codeActionProvider") is not True:
raise AssertionError(f"missing codeAction capability: {response}")
client.notify("initialized", {})
client.notify(
"textDocument/didOpen",
{"textDocument": {"uri": uri, "languageId": "viper", "version": 1, "text": text}},
)
client.wait_notification("textDocument/publishDiagnostics")
field_pos = _position(text, "field: int")
prepared = client.request(
"textDocument/prepareRename",
{"textDocument": {"uri": uri}, "position": field_pos},
).get("result")
if not isinstance(prepared, dict) or prepared.get("placeholder") != "field":
raise AssertionError(f"prepareRename failed: {prepared}")
if prepared.get("range", {}).get("start") != field_pos:
raise AssertionError(f"prepareRename range is wrong: {prepared}")
renamed = client.request(
"textDocument/rename",
{"textDocument": {"uri": uri}, "position": field_pos, "newName": "value"},
).get("result")
if not isinstance(renamed, dict):
raise AssertionError(f"rename failed: {renamed}")
edits = renamed.get("changes", {}).get(uri)
if not isinstance(edits, list) or len(edits) < 3:
raise AssertionError(f"rename produced too few edits: {renamed}")
new_text = _apply_edits(text, edits)
if " value: int" not in new_text:
raise AssertionError(f"rename did not update the declaration: {new_text}")
if "self.field" in new_text or "self.value" not in new_text:
raise AssertionError(f"rename did not update field accesses: {new_text}")
invalid = client.request(
"textDocument/rename",
{"textDocument": {"uri": uri}, "position": field_pos, "newName": "def"},
).get("result")
if invalid is not None:
raise AssertionError(f"reserved word rename should return null: {invalid}")
semantic = client.request(
"textDocument/semanticTokens/full", {"textDocument": {"uri": uri}}
).get("result")
if not isinstance(semantic, dict):
raise AssertionError(f"semanticTokens/full failed: {semantic}")
tokens = _decode_tokens(semantic.get("data", []))
keyword_pos = _position(text, "class Thing")
if not _find_token(tokens, keyword_pos["line"], keyword_pos["character"], 7):
raise AssertionError(f"missing class keyword token: {tokens[:10]}")
if not _find_token(tokens, keyword_pos["line"], keyword_pos["character"] + 6, 1):
raise AssertionError(f"missing class name token: {tokens[:10]}")
field_decl = _position(text, "field: int")
if not _find_token(tokens, field_decl["line"], field_decl["character"], 6):
raise AssertionError(f"missing property token: {tokens[:20]}")
arg_decl = _position(text, "arg: int")
if not _find_token(tokens, arg_decl["line"], arg_decl["character"], 4):
raise AssertionError(f"missing parameter token: {tokens[:20]}")
comment_pos = _position(text, "# gate5 comment")
if not _find_token(tokens, comment_pos["line"], comment_pos["character"], 8):
raise AssertionError(f"missing comment token: {tokens[:20]}")
string_pos = _position(text, '"hello"')
if not _find_token(tokens, string_pos["line"], string_pos["character"], 9):
raise AssertionError(f"missing string token: {tokens[:20]}")
number_pos = _position(text, "42")
if not _find_token(tokens, number_pos["line"], number_pos["character"], 10):
raise AssertionError(f"missing number token: {tokens[:20]}")
ranged = client.request(
"textDocument/semanticTokens/range",
{
"textDocument": {"uri": uri},
"range": {
"start": {"line": keyword_pos["line"], "character": 0},
"end": {"line": keyword_pos["line"], "character": 80},
},
},
).get("result")
if not isinstance(ranged, dict):
raise AssertionError(f"semanticTokens/range failed: {ranged}")
ranged_tokens = _decode_tokens(ranged.get("data", []))
if not ranged_tokens:
raise AssertionError(f"semanticTokens/range returned nothing: {ranged}")
for token in ranged_tokens:
if token["line"] != keyword_pos["line"]:
raise AssertionError(f"range tokens escaped the request: {token}")
actions = client.request(
"textDocument/codeAction",
{
"textDocument": {"uri": uri},
"range": {"start": {"line": 0, "character": 0}, "end": {"line": 0, "character": 0}},
"context": {"diagnostics": []},
},
).get("result")
if not isinstance(actions, list):
raise AssertionError(f"codeAction failed: {actions}")
organize = next(
(a for a in actions if a.get("kind") == "source.organizeImports"), None
)
if organize is None:
raise AssertionError(f"organize imports action missing: {actions}")
organize_edits = organize.get("edit", {}).get("changes", {}).get(uri)
if not isinstance(organize_edits, list) or not organize_edits:
raise AssertionError(f"organize imports edit is empty: {organize}")
organized = _apply_edits(text, organize_edits)
expected = (
"from .gate3_helper import helper_func\n"
"from .gate3_stub import StubThing\n"
"import stdlib\n"
)
if not organized.startswith(expected):
raise AssertionError(f"organize imports produced the wrong block: {organized[:200]}")
if organized.count("from .gate3_helper import helper_func") != 1:
raise AssertionError(f"organize imports did not deduplicate: {organized[:200]}")
client.notify("textDocument/didClose", {"textDocument": {"uri": uri}})
client.wait_notification("textDocument/publishDiagnostics")
client.request("shutdown")
code = client.close()
if code != 0:
raise AssertionError(f"exit code is {code}, expected 0")
print("rename: OK")
print("semantic-tokens: OK")
print("code-action: OK")
return 0
def main() -> int:
parser = argparse.ArgumentParser(description="ViperC LSP probe")
parser.add_argument("--selfcheck", action="store_true", help="run pure-Python framing checks")
@@ -749,6 +963,7 @@ def main() -> int:
parser.add_argument("--gate2", action="store_true", help="run syntax diagnostic checks")
parser.add_argument("--gate3", action="store_true", help="run symbol/navigation checks")
parser.add_argument("--gate4", action="store_true", help="run completion/signature checks")
parser.add_argument("--gate5", action="store_true", help="run rename/semantic/code-action checks")
parser.add_argument("--exe", help="path to viperc.exe")
args = parser.parse_args()
@@ -760,6 +975,8 @@ def main() -> int:
if not os.path.isfile(exe):
print(f"probe: executable not found: {exe}", file=sys.stderr)
return 2
if args.gate5:
return gate5(exe)
if args.gate4:
return gate4(exe)
if args.gate3: