feat(navigation): add symbol index and LSP navigation features

The changes introduce a Tcl symbol index powering LSP navigation
features across the workspace. A navigation API exposes
snapshots and update hooks, enabling goto-definition,
references, and rename using the index. Background indexing
now watches Tcl files and rebuilds the index to stay in sync.

- Add Tcl symbol index and navigation snapshot API
- Wire go-to-definition, references, and rename using the index
- Watch Tcl files and refresh the index in the background
This commit is contained in:
Christoph Brandau
2026-08-17 09:24:45 +02:00
parent f5bd79f067
commit 35a4357551
8 changed files with 1128 additions and 88 deletions
+96
View File
@@ -137,6 +137,102 @@ export function hoverCdlHandler(document: vscode.TextDocument, position: vscode.
return undefined
}
function escapeRegExp(value: string): string {
return value.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")
}
export function cdlEventAtPosition(
document: vscode.TextDocument,
position: vscode.Position
): string | undefined {
const line = document.lineAt(position.line).text
const match = /^\s*EVENT\s+([^\s{]+)/.exec(line)
if (!match) {
return undefined
}
const declarationStart = match.index
const eventEnd = line.indexOf(match[1], match.index) + match[1].length
if (position.character < declarationStart || position.character > eventEnd) {
return undefined
}
return match[1]
}
export async function definitionCdlEventHandler(
document: vscode.TextDocument,
position: vscode.Position,
token: vscode.CancellationToken
): Promise<vscode.Location[] | undefined> {
const eventName = cdlEventAtPosition(document, position)
if (!eventName) {
return undefined
}
const handlerName = `MOM_${eventName}`
try {
const symbols = await vscode.commands.executeCommand<vscode.SymbolInformation[]>(
"vscode.executeWorkspaceSymbolProvider",
handlerName
)
const indexedLocations = (symbols || [])
.filter(
(symbol) =>
symbol.kind === vscode.SymbolKind.Function &&
(symbol.name === handlerName ||
symbol.name.endsWith(`::${handlerName}`))
)
.map((symbol) => symbol.location)
if (indexedLocations.length > 0) {
return indexedLocations
}
} catch {
// The Tcl language server may still be starting; use the file fallback below.
}
const declaration = new RegExp(
`^\\s*proc\\s+(?:::)?${escapeRegExp(handlerName)}(?=\\s|\\{)`
)
const tclFiles = await vscode.workspace.findFiles(
"**/*.tcl",
"**/{.git,.nox,.venv,dist,node_modules,out}/**"
)
const locations: vscode.Location[] = []
for (const uri of tclFiles) {
if (token.isCancellationRequested) {
return undefined
}
let tclDocument: vscode.TextDocument
try {
tclDocument = await vscode.workspace.openTextDocument(uri)
} catch {
continue
}
for (let lineNumber = 0; lineNumber < tclDocument.lineCount; lineNumber++) {
const line = tclDocument.lineAt(lineNumber).text
const match = declaration.exec(line)
if (!match) {
continue
}
const start = line.indexOf(handlerName, match.index)
locations.push(
new vscode.Location(
uri,
new vscode.Range(
lineNumber,
start,
lineNumber,
start + handlerName.length
)
)
)
}
}
return locations.length > 0 ? locations : undefined
}
export function tclDocumentSymbolProvider(document: vscode.TextDocument): vscode.DocumentSymbol[] {
const symbols: vscode.DocumentSymbol[] = []
const lines = document.getText().split("\n")
+15 -4
View File
@@ -2,7 +2,7 @@
// Licensed under the MIT License.
import * as fsapi from "fs-extra"
import { Disposable, env, LogOutputChannel } from "vscode"
import { Disposable, env, LogOutputChannel, workspace } from "vscode"
import { State } from "vscode-languageclient"
import {
LanguageClient,
@@ -24,6 +24,13 @@ import { isVirtualWorkspace } from "./vscodeapi"
export type IInitOptions = { settings: ISettings[]; globalSettings: ISettings }
let _disposables: Disposable[] = []
export function disposeServerResources(): void {
_disposables.forEach((disposable) => disposable.dispose())
_disposables = []
}
async function createServer(
settings: ISettings,
serverId: string,
@@ -63,6 +70,7 @@ async function createServer(
}
// Options to control the language client
const tclFileWatcher = workspace.createFileSystemWatcher("**/*.tcl")
const clientOptions: LanguageClientOptions = {
// Register the server for python documents
documentSelector: isVirtualWorkspace()
@@ -76,13 +84,16 @@ async function createServer(
outputChannel: outputChannel,
traceOutputChannel: outputChannel,
revealOutputChannelOn: RevealOutputChannelOn.Never,
synchronize: {
fileEvents: tclFileWatcher
},
initializationOptions
}
_disposables.push(tclFileWatcher)
return new LanguageClient(serverId, serverName, serverOptions, clientOptions)
}
let _disposables: Disposable[] = []
export async function restartServer(
serverId: string,
serverName: string,
@@ -92,8 +103,7 @@ export async function restartServer(
if (lsClient) {
traceInfo(`Server: Stop requested`)
await lsClient.stop()
_disposables.forEach((d) => d.dispose())
_disposables = []
disposeServerResources()
}
const projectRoot = await getProjectRoot()
const workspaceSetting = await getWorkspaceSettings(serverId, projectRoot, true)
@@ -122,6 +132,7 @@ export async function restartServer(
await newLSClient.start()
} catch (ex) {
traceError(`Server: Start failed: ${ex}`)
disposeServerResources()
return undefined
}
+15 -3
View File
@@ -13,7 +13,8 @@ import {
isFirstLineMachine,
diagnosticHandler,
cdlDocumentSymbolProvider,
defDocumentSymbolProvider
defDocumentSymbolProvider,
definitionCdlEventHandler
} from "./common/handlers"
import { registerLogger, traceError, traceLog, traceVerbose } from "./common/log/logging"
import {
@@ -23,7 +24,7 @@ import {
onDidChangePythonInterpreter,
resolveInterpreter
} from "./common/python"
import { restartServer } from "./common/server"
import { disposeServerResources, restartServer } from "./common/server"
import { checkIfConfigurationChanged, getInterpreterFromSetting } from "./common/settings"
import { loadServerDefaults } from "./common/setup"
import { getLSClientTraceLevel } from "./common/utilities"
@@ -177,6 +178,16 @@ export async function activate(context: vscode.ExtensionContext) {
)
context.subscriptions.push(hoverCdlProvider)
const definitionCdlEventProvider = vscode.languages.registerDefinitionProvider(
{ scheme: "file", language: "cdl" },
{
provideDefinition(document, position, token) {
return definitionCdlEventHandler(document, position, token)
}
}
)
context.subscriptions.push(definitionCdlEventProvider)
const formatDefProvider = vscode.languages.registerDocumentFormattingEditProvider(
{ scheme: "file", language: "def" },
{
@@ -246,7 +257,8 @@ export async function activate(context: vscode.ExtensionContext) {
export function deactivate(): Thenable<void> | undefined {
if (!client) {
disposeServerResources()
return undefined
}
return client.stop()
return client.stop().finally(disposeServerResources)
}
+192 -81
View File
@@ -45,8 +45,14 @@ from common.load_data import standard_items
from tools.folding_ranges import build_folding_ranges
from tools.semantic_tokens import _Highlighter, TOKEN_TYPES, TokenModifier
from tools.inlay_hint import InlayHintGenerator
from tools.navigation import (
SymbolIdentity,
definition_identities,
matching_occurrences,
symbol_at_position,
workspace_symbols,
)
from tools.signature_help import build_signature_help
from tools.file_sourcing import get_all_psc_files, read_psc_file
from lsp_tclserver import TclLanguageServer
from pygls.workspace.text_document import TextDocument
@@ -60,6 +66,12 @@ LSP_SERVER = TclLanguageServer(
name="NX Postprocessor Support", version="0.0.1", max_workers=MAX_WORKERS
)
BUILTIN_PROC_NAMES = {
item.label
for item in standard_items.tcl_keyword_list + standard_items.nx_procs
}
BUILTIN_VARIABLE_NAMES = {item.label for item in standard_items.nx_variables}
# **********************************************************
# Tool specific code goes below this.
# **********************************************************
@@ -166,6 +178,16 @@ def did_rename_files(params: lsp.RenameFilesParams) -> None:
_index_tcl_file_from_disk(new_index_path.as_uri())
@LSP_SERVER.feature(lsp.WORKSPACE_DID_CHANGE_WATCHED_FILES)
def did_change_watched_files(params: lsp.DidChangeWatchedFilesParams) -> None:
"""Keep indexes for closed Tcl files synchronized with disk changes."""
for change in params.changes:
if change.type == lsp.FileChangeType.Deleted:
LSP_SERVER.remove_file_state(change.uri)
else:
_index_tcl_file_from_disk(change.uri)
@LSP_SERVER.feature(
lsp.TEXT_DOCUMENT_DIAGNOSTIC,
lsp.DiagnosticOptions(
@@ -448,74 +470,153 @@ def hover(params: lsp.HoverParams) -> lsp.Hover:
@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DEFINITION)
def goto_definition(params: lsp.DefinitionParams):
"""Provide go-to-definition locations for Tcl procs.
Strategy:
- Find the token under the cursor.
- If it matches a custom proc collected in proc_signatures, locate its declaration
by searching the current document first, then other indexed files.
- Return a Location pointing to the proc name in its declaration line.
"""
doc = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
pos = params.position
try:
line = doc.lines[pos.line]
except IndexError:
"""Resolve Tcl proc and variable definitions through the symbol index."""
context = _navigation_context(params.text_document.uri, params.position)
if context is None:
return None
# Identify token under cursor
token = None
for m in re.finditer(r"\b\w+\b", line):
if m.start() <= pos.character <= m.end():
token = m.group(0)
break
if not token:
indexes, definitions, _, identity = context
locations = [
lsp.Location(uri=index.uri, range=occurrence.range)
for index, occurrence in matching_occurrences(
identity, indexes, definitions
)
if occurrence.is_definition
]
return _sorted_locations(locations) or None
def _navigation_context(uri: str, position: lsp.Position):
indexes = LSP_SERVER.navigation_snapshot()
filepath = str(pathlib.Path(uris.to_fs_path(uri)))
index = indexes.get(filepath)
if index is None:
document = LSP_SERVER.workspace.get_text_document(uri)
LSP_SERVER.update_poco_completion_for_file(document)
indexes = LSP_SERVER.navigation_snapshot()
index = indexes.get(filepath)
if index is None:
return None
# Helper to search a single source text for a proc declaration
def find_decl_in_source(source_text: str, uri: str) -> Optional[lsp.Location]:
lines = source_text.split("\n")
pattern = re.compile(r"^\s*proc\s+" + re.escape(token) + r"\b")
for i, ln in enumerate(lines):
m = pattern.match(ln)
if m:
start_char = ln.find(token)
if start_char < 0:
start_char = max(m.end() - len(token), 0)
start = lsp.Position(i, start_char)
end = lsp.Position(i, start_char + len(token))
return lsp.Location(uri=uri, range=lsp.Range(start=start, end=end))
definitions = definition_identities(indexes)
result = symbol_at_position(index, position, definitions)
if result is None:
return None
occurrence, identity = result
return indexes, definitions, occurrence, identity
def _sorted_locations(locations: list[lsp.Location]) -> list[lsp.Location]:
return sorted(
locations,
key=lambda location: (
location.uri,
location.range.start.line,
location.range.start.character,
),
)
@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_REFERENCES)
def references(params: lsp.ReferenceParams) -> list[lsp.Location]:
context = _navigation_context(params.text_document.uri, params.position)
if context is None:
return []
indexes, definitions, _, identity = context
locations = [
lsp.Location(uri=index.uri, range=occurrence.range)
for index, occurrence in matching_occurrences(
identity, indexes, definitions
)
if params.context.include_declaration or not occurrence.is_definition
]
return _sorted_locations(locations)
def _is_renamable(
identity: SymbolIdentity,
indexes,
definitions: set[SymbolIdentity],
) -> bool:
if identity not in definitions or identity.kind not in {"proc", "variable"}:
return False
basename = identity.name.rsplit("::", 1)[-1]
if identity.kind == "proc":
if basename in BUILTIN_PROC_NAMES:
return False
definition_count = sum(
occurrence.is_definition and occurrence.identity == identity
for index in indexes.values()
for occurrence in index.occurrences
)
return definition_count == 1
return basename not in BUILTIN_VARIABLE_NAMES
@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_PREPARE_RENAME)
def prepare_rename(params: lsp.PrepareRenameParams):
context = _navigation_context(params.text_document.uri, params.position)
if context is None:
return None
# 1) Search in current document
loc = find_decl_in_source(doc.source, doc.uri)
if loc:
return loc
indexes, definitions, occurrence, identity = context
if not _is_renamable(identity, indexes, definitions):
return None
return lsp.PrepareRenameResult_Type1(
range=occurrence.range, placeholder=occurrence.placeholder
)
# 2) Search in indexed files from proc_signatures
# Build list of candidate files that declare this token as a proc
candidate_files: list[str] = []
_, proc_signatures, _ = LSP_SERVER.index_snapshot()
for file_path, procs in proc_signatures.items():
if token in procs:
candidate_files.append(file_path)
for fp in candidate_files:
uri = pathlib.Path(fp).as_uri()
# Try to get from workspace if available; else read from disk
try:
other_doc = LSP_SERVER.workspace.get_text_document(uri)
source = other_doc.source
except Exception:
try:
source = pathlib.Path(fp).read_text(encoding="utf-8")
except Exception:
continue
loc = find_decl_in_source(source, uri)
if loc:
return loc
@LSP_SERVER.feature(
lsp.TEXT_DOCUMENT_RENAME,
lsp.RenameOptions(prepare_provider=True),
)
def rename(params: lsp.RenameParams) -> lsp.WorkspaceEdit | None:
if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", params.new_name):
return None
return None
context = _navigation_context(params.text_document.uri, params.position)
if context is None:
return None
indexes, definitions, _, identity = context
if not _is_renamable(identity, indexes, definitions):
return None
changes: dict[str, list[lsp.TextEdit]] = {}
seen = set()
for index, occurrence in matching_occurrences(identity, indexes, definitions):
key = (
index.uri,
occurrence.range.start.line,
occurrence.range.start.character,
occurrence.range.end.line,
occurrence.range.end.character,
)
if key in seen:
continue
seen.add(key)
changes.setdefault(index.uri, []).append(
lsp.TextEdit(range=occurrence.range, new_text=params.new_name)
)
for edits in changes.values():
edits.sort(
key=lambda edit: (
edit.range.start.line,
edit.range.start.character,
),
reverse=True,
)
return lsp.WorkspaceEdit(changes=changes)
@LSP_SERVER.feature(lsp.WORKSPACE_SYMBOL)
def workspace_symbol(params: lsp.WorkspaceSymbolParams):
return workspace_symbols(LSP_SERVER.navigation_snapshot(), params.query)
# **********************************************************
@@ -590,6 +691,9 @@ def initialize(params: lsp.InitializeParams) -> lsp.InitializeResult:
legend=semantic_tokens_legend, full=True, range=False
),
definition_provider=True,
references_provider=True,
rename_provider=lsp.RenameOptions(prepare_provider=True),
workspace_symbol_provider=True,
)
)
@@ -602,28 +706,35 @@ def initialized(_params: lsp.InitializedParams):
try:
root = LSP_SERVER.workspace.root_path
log_to_output("Background indexing started...")
psc_files = get_all_psc_files(pathlib.Path(root))
for psc_file in psc_files:
poco_files = read_psc_file(psc_file)
for sourced_layer in poco_files:
file_root = pathlib.Path(root).joinpath(
sourced_layer.subfolder if sourced_layer.subfolder else ""
root_path = pathlib.Path(root)
skipped_directories = {
".git",
".nox",
".venv",
"dist",
"node_modules",
"out",
}
tcl_files = (
path
for path in root_path.rglob("*.tcl")
if not any(
part.casefold() in skipped_directories
for part in path.relative_to(root_path).parts[:-1]
)
)
for filepath in sorted(tcl_files, key=lambda path: str(path).casefold()):
try:
document = TextDocument(
uri=filepath.as_uri(), language_id="tcl"
)
for tcl_file in sourced_layer.files:
filepath = pathlib.Path(file_root).joinpath(f"{tcl_file}.tcl")
if not filepath.exists():
continue
try:
document = TextDocument(
uri=filepath.as_uri(), language_id="tcl"
)
LSP_SERVER.update_poco_completion_for_file(
document,
cache_tree=False,
require_file_exists=True,
)
except Exception as error:
log_to_output(f"Fehler beim Parsen von {filepath}: {error}")
LSP_SERVER.update_poco_completion_for_file(
document,
cache_tree=False,
require_file_exists=True,
)
except Exception as error:
log_to_output(f"Fehler beim Parsen von {filepath}: {error}")
log_to_output("Background indexing completed.")
except Exception as e:
log_to_output(f"Background indexing failed: {e}")
+14
View File
@@ -15,6 +15,7 @@ from tclint.violations import Violation
from tools import checks, parser
from tools.completion_items import CompletionCollector
from tools.formatter import NxFormatter as Formatter
from tools.navigation import FileSymbolIndex, build_file_symbol_index
from tools.proc_docs import build_proc_docs
@@ -31,6 +32,7 @@ class TclLanguageServer(server.LanguageServer):
self.poco_completion: dict = {}
self.proc_signatures: dict = {}
self.proc_docs: dict = {}
self.navigation_indexes: dict[str, FileSymbolIndex] = {}
# Cache: (uri, version) -> (tree, violations)
self._ast_cache = {}
self._parser_lock = threading.RLock()
@@ -115,6 +117,11 @@ class TclLanguageServer(server.LanguageServer):
with self._index_lock:
return self.diagnostics.get(uri)
def navigation_snapshot(self) -> dict[str, FileSymbolIndex]:
"""Return an immutable snapshot of the workspace symbol indexes."""
with self._index_lock:
return dict(self.navigation_indexes)
def _begin_index_update(self, filepath: str, version: int | None) -> int | None:
with self._index_lock:
indexed_version = self._index_versions.get(filepath)
@@ -140,6 +147,7 @@ class TclLanguageServer(server.LanguageServer):
self.poco_completion.pop(filepath, None)
self.proc_signatures.pop(filepath, None)
self.proc_docs.pop(filepath, None)
self.navigation_indexes.pop(filepath, None)
def indexed_paths_under_uri(self, uri: str) -> list[pathlib.Path]:
target = pathlib.Path(uris.to_fs_path(uri))
@@ -147,6 +155,7 @@ class TclLanguageServer(server.LanguageServer):
indexed_paths = set(self.poco_completion)
indexed_paths.update(self.proc_signatures)
indexed_paths.update(self.proc_docs)
indexed_paths.update(self.navigation_indexes)
indexed_paths.update(self._index_tokens)
return [
pathlib.Path(path)
@@ -163,6 +172,7 @@ class TclLanguageServer(server.LanguageServer):
self.poco_completion,
self.proc_signatures,
self.proc_docs,
self.navigation_indexes,
self._index_tokens,
self._index_versions,
):
@@ -212,6 +222,9 @@ class TclLanguageServer(server.LanguageServer):
)
tree.accept(collector, recurse=True)
docs = build_proc_docs(tree, document.source)
navigation_index = build_file_symbol_index(
filepath, document.uri, tree
)
except Exception as e:
logging.debug(f"Error parsing {filepath}: {e}")
self._discard_index_update(filepath, token)
@@ -227,6 +240,7 @@ class TclLanguageServer(server.LanguageServer):
self.poco_completion[filepath] = list(collector.custom_functions)
self.proc_signatures[filepath] = dict(collector.proc_signatures)
self.proc_docs[filepath] = docs
self.navigation_indexes[filepath] = navigation_index
return True
def format(
+519
View File
@@ -0,0 +1,519 @@
from __future__ import annotations
from dataclasses import dataclass
import lsprotocol.types as lsp
from tclint.syntax_tree import BareWord, Command, List, Node, Script, VarSub
ROOT_NAMESPACE = "::"
@dataclass(frozen=True)
class SymbolIdentity:
kind: str
name: str
scope: str | None = None
@dataclass(frozen=True)
class SymbolOccurrence:
identity: SymbolIdentity
range: lsp.Range
placeholder: str
is_definition: bool = False
symbol_kind: lsp.SymbolKind = lsp.SymbolKind.Variable
container_name: str | None = None
fallback_identity: SymbolIdentity | None = None
@dataclass(frozen=True)
class FileSymbolIndex:
path: str
uri: str
occurrences: tuple[SymbolOccurrence, ...]
@dataclass(frozen=True)
class _Scope:
filepath: str
namespace: str = ROOT_NAMESPACE
proc_name: str | None = None
global_variables: tuple[tuple[str, str], ...] = ()
namespace_variables: tuple[tuple[str, str], ...] = ()
def _without_array_index(name: str) -> str:
return name.split("(", 1)[0]
def _basename(name: str) -> str:
return _without_array_index(name).rsplit("::", 1)[-1]
def _qualify(name: str, namespace: str) -> str:
name = _without_array_index(name)
if name.startswith("::"):
return name
if namespace == ROOT_NAMESPACE:
return f"::{name}"
return f"{namespace}::{name}"
def _namespace_of(qualified_name: str) -> str:
parent = qualified_name.rsplit("::", 1)[0]
return parent or ROOT_NAMESPACE
def _display_name(identity: SymbolIdentity) -> str:
if identity.kind == "variable" and identity.scope is not None:
return identity.name
return identity.name.removeprefix("::")
def _container_name(identity: SymbolIdentity) -> str | None:
if identity.scope is not None:
_, _, proc_name = identity.scope.partition("::proc::")
return proc_name.removeprefix("::") or None
qualified = identity.name.removeprefix("::")
if "::" not in qualified:
return None
return qualified.rsplit("::", 1)[0]
def _static_contents(node: Node | None) -> str | None:
value = getattr(node, "contents", None)
return value if isinstance(value, str) else None
def _name_range(node: Node, raw_name: str, *, variable_sub: bool = False) -> lsp.Range:
if variable_sub:
line, column = node.pos
column += 2 if getattr(node, "braced", False) else 1
else:
position = getattr(node, "contents_pos", None) or node.pos
line, column = position
normalized = _without_array_index(raw_name)
prefix_length = normalized.rfind("::") + 2 if "::" in normalized else 0
start_character = column - 1 + prefix_length
name = normalized[prefix_length:]
return lsp.Range(
start=lsp.Position(line=line - 1, character=start_character),
end=lsp.Position(line=line - 1, character=start_character + len(name)),
)
def _proc_identity(raw_name: str, namespace: str) -> SymbolIdentity:
return SymbolIdentity(kind="proc", name=_qualify(raw_name, namespace))
def _proc_fallback(raw_name: str, namespace: str) -> SymbolIdentity | None:
if raw_name.startswith("::") or "::" in raw_name or namespace == ROOT_NAMESPACE:
return None
return SymbolIdentity(kind="proc", name=_qualify(raw_name, ROOT_NAMESPACE))
def _variable_identity(raw_name: str, scope: _Scope) -> SymbolIdentity:
normalized = _without_array_index(raw_name)
if normalized.startswith("::"):
return SymbolIdentity(kind="variable", name=normalized)
if "::" in normalized:
return SymbolIdentity(
kind="variable", name=_qualify(normalized, scope.namespace)
)
if scope.proc_name is None:
return SymbolIdentity(
kind="variable", name=_qualify(normalized, scope.namespace)
)
for alias, target in scope.global_variables + scope.namespace_variables:
if normalized == alias:
return SymbolIdentity(kind="variable", name=target)
return SymbolIdentity(
kind="variable",
name=normalized,
scope=f"{scope.filepath}::proc::{scope.proc_name}",
)
def _variable_command_nodes(command: Command) -> list[tuple[Node, bool]]:
routine = _static_contents(command.routine)
if routine == "set" and command.args:
return [(command.args[0], len(command.args) >= 2)]
if routine in {"incr", "append", "lappend"} and command.args:
return [(command.args[0], True)]
if routine == "lset" and command.args:
return [(command.args[0], False)]
if routine == "unset":
return [
(argument, False)
for argument in command.args
if not (_static_contents(argument) or "").startswith("-")
]
if routine == "array" and len(command.args) >= 2:
return [
(command.args[1], _static_contents(command.args[0]) == "set")
]
if routine == "dict" and len(command.args) >= 2:
subcommand = _static_contents(command.args[0])
if subcommand in {"set", "unset", "append", "incr", "lappend", "update", "with"}:
return [
(
command.args[1],
subcommand in {"set", "append", "incr", "lappend"},
)
]
return []
def _binding_nodes(node: Node) -> list[Node]:
if isinstance(node, List):
return list(node.children)
return [node]
def _variable_binding_nodes(command: Command) -> list[Node]:
routine = _static_contents(command.routine)
if routine in {"foreach", "lmap"} and len(command.args) >= 3:
return [
variable
for variable_list in command.args[:-1:2]
for variable in _binding_nodes(variable_list)
]
if routine == "lassign" and len(command.args) >= 2:
return list(command.args[1:])
if routine == "catch" and len(command.args) >= 2:
return list(command.args[1:3])
if (
routine == "dict"
and command.args
and _static_contents(command.args[0]) == "update"
):
return list(command.args[3:-1:2])
return []
def _variable_declaration_nodes(command: Command) -> list[Node]:
routine = _static_contents(command.routine)
if routine == "global":
return list(command.args)
if routine == "variable":
return list(command.args[::2])
return []
def _scan_proc_imports(
node: Node, namespace: str
) -> tuple[dict[str, str], dict[str, str]]:
global_variables: dict[str, str] = {}
namespace_variables: dict[str, str] = {}
def walk(current: Node) -> None:
if isinstance(current, Command):
routine = _static_contents(current.routine)
if routine == "proc":
return
if routine == "global":
for argument in current.args:
name = _static_contents(argument)
if name:
global_variables[_basename(name)] = _qualify(
name, ROOT_NAMESPACE
)
elif routine == "variable":
for argument in current.args[::2]:
name = _static_contents(argument)
if name:
namespace_variables[_basename(name)] = _qualify(
name, namespace
)
for child in getattr(current, "children", []):
walk(child)
walk(node)
return global_variables, namespace_variables
def build_file_symbol_index(
filepath: str, uri: str, tree: Node
) -> FileSymbolIndex:
occurrences: list[SymbolOccurrence] = []
def add_proc(
node: Node,
raw_name: str,
scope: _Scope,
*,
is_definition: bool,
) -> None:
identity = _proc_identity(raw_name, scope.namespace)
occurrences.append(
SymbolOccurrence(
identity=identity,
fallback_identity=(
None
if is_definition
else _proc_fallback(raw_name, scope.namespace)
),
range=_name_range(node, raw_name),
placeholder=_basename(raw_name),
is_definition=is_definition,
symbol_kind=lsp.SymbolKind.Function,
container_name=_container_name(identity),
)
)
def add_variable(
node: Node,
raw_name: str,
scope: _Scope,
*,
is_definition: bool,
variable_sub: bool = False,
identity: SymbolIdentity | None = None,
) -> None:
symbol_identity = identity or _variable_identity(raw_name, scope)
occurrences.append(
SymbolOccurrence(
identity=symbol_identity,
range=_name_range(node, raw_name, variable_sub=variable_sub),
placeholder=_basename(raw_name),
is_definition=is_definition,
symbol_kind=lsp.SymbolKind.Variable,
container_name=_container_name(symbol_identity),
)
)
def walk_embedded(node: Node, scope: _Scope) -> None:
if isinstance(node, Script):
walk_script(node, scope)
return
if isinstance(node, Command):
walk_command(node, scope)
return
if isinstance(node, VarSub):
raw_name = getattr(node, "value", None)
if isinstance(raw_name, str):
add_variable(
node,
raw_name,
scope,
is_definition=False,
variable_sub=True,
)
for child in getattr(node, "children", []):
walk_embedded(child, scope)
def walk_proc(command: Command, scope: _Scope) -> None:
if len(command.args) < 3:
return
raw_name = _static_contents(command.args[0])
body = command.args[2]
if raw_name is None or not isinstance(body, Script):
return
add_proc(command.args[0], raw_name, scope, is_definition=True)
proc_identity = _proc_identity(raw_name, scope.namespace)
proc_namespace = _namespace_of(proc_identity.name)
global_variables, namespace_variables = _scan_proc_imports(
body, proc_namespace
)
proc_scope = _Scope(
filepath=filepath,
namespace=proc_namespace,
proc_name=proc_identity.name,
global_variables=tuple(sorted(global_variables.items())),
namespace_variables=tuple(sorted(namespace_variables.items())),
)
parameters = command.args[1]
for parameter in getattr(parameters, "children", []):
parameter_node = parameter
if isinstance(parameter, List) and parameter.children:
parameter_node = parameter.children[0]
parameter_name = _static_contents(parameter_node)
if parameter_name:
add_variable(
parameter_node,
parameter_name,
proc_scope,
is_definition=True,
)
walk_script(body, proc_scope)
def walk_namespace(command: Command, scope: _Scope) -> bool:
if len(command.args) < 3 or _static_contents(command.args[0]) != "eval":
return False
raw_name = _static_contents(command.args[1])
body = command.args[2]
if raw_name is None or not isinstance(body, Script):
return False
namespace = _qualify(raw_name, scope.namespace)
identity = SymbolIdentity(kind="namespace", name=namespace)
occurrences.append(
SymbolOccurrence(
identity=identity,
range=_name_range(command.args[1], raw_name),
placeholder=_basename(raw_name),
is_definition=True,
symbol_kind=lsp.SymbolKind.Namespace,
container_name=_container_name(identity),
)
)
walk_script(
body,
_Scope(filepath=filepath, namespace=namespace),
)
return True
def walk_command(command: Command, scope: _Scope) -> None:
routine = _static_contents(command.routine)
if routine == "proc":
walk_proc(command, scope)
return
if routine == "namespace" and walk_namespace(command, scope):
return
if routine:
add_proc(command.routine, routine, scope, is_definition=False)
declaration_nodes = _variable_declaration_nodes(command)
declaration_ids = {id(node) for node in declaration_nodes}
for node in declaration_nodes:
raw_name = _static_contents(node)
if raw_name:
is_definition = routine == "variable" and scope.proc_name is None
if routine == "global":
identity = SymbolIdentity(
kind="variable",
name=_qualify(raw_name, ROOT_NAMESPACE),
)
else:
identity = SymbolIdentity(
kind="variable",
name=_qualify(raw_name, scope.namespace),
)
add_variable(
node,
raw_name,
scope,
is_definition=is_definition,
identity=identity,
)
for node, is_definition in _variable_command_nodes(command):
if id(node) in declaration_ids:
continue
raw_name = _static_contents(node)
if raw_name:
add_variable(
node,
raw_name,
scope,
is_definition=is_definition,
)
for node in _variable_binding_nodes(command):
raw_name = _static_contents(node)
if raw_name:
add_variable(node, raw_name, scope, is_definition=True)
for argument in command.args:
walk_embedded(argument, scope)
def walk_script(script: Node, scope: _Scope) -> None:
for child in getattr(script, "children", []):
walk_embedded(child, scope)
walk_script(tree, _Scope(filepath=filepath))
return FileSymbolIndex(path=filepath, uri=uri, occurrences=tuple(occurrences))
def definition_identities(indexes: dict[str, FileSymbolIndex]) -> set[SymbolIdentity]:
return {
occurrence.identity
for index in indexes.values()
for occurrence in index.occurrences
if occurrence.is_definition
}
def resolve_identity(
occurrence: SymbolOccurrence, definitions: set[SymbolIdentity]
) -> SymbolIdentity:
if occurrence.identity in definitions or occurrence.fallback_identity is None:
return occurrence.identity
if occurrence.fallback_identity in definitions:
return occurrence.fallback_identity
return occurrence.identity
def symbol_at_position(
index: FileSymbolIndex,
position: lsp.Position,
definitions: set[SymbolIdentity],
) -> tuple[SymbolOccurrence, SymbolIdentity] | None:
for occurrence in index.occurrences:
start = occurrence.range.start
end = occurrence.range.end
if (
position.line == start.line == end.line
and start.character <= position.character < end.character
):
return occurrence, resolve_identity(occurrence, definitions)
return None
def matching_occurrences(
identity: SymbolIdentity,
indexes: dict[str, FileSymbolIndex],
definitions: set[SymbolIdentity],
) -> list[tuple[FileSymbolIndex, SymbolOccurrence]]:
matches = []
for index in indexes.values():
for occurrence in index.occurrences:
if resolve_identity(occurrence, definitions) == identity:
matches.append((index, occurrence))
return matches
def workspace_symbols(
indexes: dict[str, FileSymbolIndex], query: str
) -> list[lsp.SymbolInformation]:
query = query.casefold()
results = []
seen = set()
for index in indexes.values():
for occurrence in index.occurrences:
identity = occurrence.identity
if not occurrence.is_definition:
continue
if identity.kind == "variable" and identity.scope is not None:
continue
name = _display_name(identity)
if query and query not in name.casefold():
continue
key = identity
if key in seen:
continue
seen.add(key)
results.append(
lsp.SymbolInformation(
name=name,
kind=occurrence.symbol_kind,
location=lsp.Location(uri=index.uri, range=occurrence.range),
container_name=occurrence.container_name,
)
)
return sorted(results, key=lambda symbol: symbol.name.casefold())
@@ -51,6 +51,7 @@ def test_duplicate_proc_stays_indexed_when_other_file_is_removed(tmp_path: Path)
assert first.path not in completions
assert first.path not in signatures
assert first.path not in docs
assert first.path not in server.navigation_snapshot()
assert "shared" in signatures[second.path]
assert server.diagnostic_snapshot(first.uri) is None
@@ -88,6 +89,7 @@ def test_delete_and_rename_notifications_update_index(tmp_path: Path, monkeypatc
)
_, signatures, _ = server.index_snapshot()
assert deleted.path not in signatures
assert deleted.path not in server.navigation_snapshot()
old_path = tmp_path / "old.tcl"
new_path = tmp_path / "new.tcl"
@@ -112,6 +114,10 @@ def test_delete_and_rename_notifications_update_index(tmp_path: Path, monkeypatc
if server.paths_equal(indexed_path, new_path)
)
assert "renamed_proc" in renamed_signatures
assert any(
server.paths_equal(indexed_path, new_path)
for indexed_path in server.navigation_snapshot()
)
def test_parallel_file_indexing_keeps_every_file(tmp_path: Path):
@@ -0,0 +1,271 @@
import sys
from pathlib import Path
THIS_DIR = Path(__file__).parent
SRC_DIR = THIS_DIR.parent.parent / "src"
if str(SRC_DIR) not in sys.path:
sys.path.insert(0, str(SRC_DIR))
import lsprotocol.types as lsp # type: ignore
from pygls.workspace.text_document import TextDocument
import lsp_server
from lsp_tclserver import TclLanguageServer
from tools.navigation import (
SymbolIdentity,
build_file_symbol_index,
definition_identities,
matching_occurrences,
symbol_at_position,
workspace_symbols,
)
from tools.parser import CustomParser
def _index(path: Path, source: str):
return build_file_symbol_index(
str(path), path.as_uri(), CustomParser().parse(source)
)
def _document(path: Path, source: str) -> TextDocument:
return TextDocument(
uri=path.as_uri(),
source=source,
version=1,
language_id="tcl",
)
def _position(source: str, token: str, occurrence: int = 0) -> lsp.Position:
offset = -1
for _ in range(occurrence + 1):
offset = source.index(token, offset + 1)
before = source[:offset]
return lsp.Position(
line=before.count("\n"),
character=offset - (before.rfind("\n") + 1),
)
def _range_text(source: str, range_: lsp.Range) -> str:
assert range_.start.line == range_.end.line
line = source.splitlines()[range_.start.line]
return line[range_.start.character : range_.end.character]
def test_proc_references_respect_namespaces_and_root_fallback(tmp_path: Path):
first_source = """proc shared {value} { return $value }
namespace eval shop {
proc shared {value} { return $value }
proc call {} { shared 1 }
}
"""
second_source = """shared 2
namespace eval shop { shared 3 }
::shop::shared 4
"""
first = _index(tmp_path / "first.tcl", first_source)
second = _index(tmp_path / "second.tcl", second_source)
indexes = {first.path: first, second.path: second}
definitions = definition_identities(indexes)
root = SymbolIdentity(kind="proc", name="::shared")
namespaced = SymbolIdentity(kind="proc", name="::shop::shared")
assert len(matching_occurrences(root, indexes, definitions)) == 2
assert len(matching_occurrences(namespaced, indexes, definitions)) == 4
def test_local_variable_identity_does_not_leak_between_procs(tmp_path: Path):
source = """proc first {} {
set value 1
puts $value
}
proc second {} {
set value 2
puts $value
}
"""
index = _index(tmp_path / "locals.tcl", source)
indexes = {index.path: index}
definitions = definition_identities(indexes)
position = _position(source, "$value")
result = symbol_at_position(
index,
lsp.Position(position.line, position.character + 1),
definitions,
)
assert result is not None
_, identity = result
matches = matching_occurrences(identity, indexes, definitions)
assert len(matches) == 2
assert all(match.identity.scope and "::first" in match.identity.scope for _, match in matches)
def test_foreach_binding_can_be_renamed_without_touching_other_proc(
tmp_path: Path,
):
source = """proc first {items} {
foreach item $items { puts $item }
}
proc second {items} {
foreach item $items { puts $item }
}
"""
index = _index(tmp_path / "foreach.tcl", source)
indexes = {index.path: index}
definitions = definition_identities(indexes)
position = _position(source, "item", occurrence=1)
result = symbol_at_position(index, position, definitions)
assert result is not None
_, identity = result
matches = matching_occurrences(identity, indexes, definitions)
assert identity in definitions
assert len(matches) == 2
assert all("::first" in (occurrence.identity.scope or "") for _, occurrence in matches)
def test_variable_ranges_preserve_qualifiers_and_tcl_substitution(tmp_path: Path):
source = """namespace eval shop {
variable value 0
proc use {} {
variable value
set value 1
puts ${value}
}
}
set ::shop::value 2
"""
index = _index(tmp_path / "variables.tcl", source)
indexes = {index.path: index}
definitions = definition_identities(indexes)
identity = SymbolIdentity(kind="variable", name="::shop::value")
matches = matching_occurrences(identity, indexes, definitions)
assert len(matches) == 5
assert all(_range_text(source, occurrence.range) == "value" for _, occurrence in matches)
def test_qualified_proc_body_and_variable_import_use_declared_namespace(
tmp_path: Path,
):
source = """namespace eval current {
proc ::other::use {} {
variable value
puts $value
variable ::external::setting
puts $setting
}
}
namespace eval other { variable value 1 }
namespace eval external { variable setting 2 }
"""
index = _index(tmp_path / "qualified.tcl", source)
indexes = {index.path: index}
definitions = definition_identities(indexes)
other_value = SymbolIdentity(kind="variable", name="::other::value")
external_setting = SymbolIdentity(
kind="variable", name="::external::setting"
)
assert len(matching_occurrences(other_value, indexes, definitions)) == 3
assert len(matching_occurrences(external_setting, indexes, definitions)) == 3
def test_workspace_symbols_include_procs_namespaces_and_global_variables(tmp_path: Path):
source = """set globalValue 1
set globalValue 2
proc rootProc {} { return }
namespace eval shop { proc namespacedProc {} { return } }
"""
index = _index(tmp_path / "symbols.tcl", source)
symbols = workspace_symbols({index.path: index}, "")
names = [symbol.name for symbol in symbols]
assert names.count("globalValue") == 1
assert "rootProc" in names
assert "shop" in names
assert "shop::namespacedProc" in names
def test_lsp_references_definition_rename_and_workspace_symbols(
tmp_path: Path, monkeypatch
):
declaration_source = "proc customProc {value} { return $value }\n"
usage_source = "set result [customProc 1]\n"
declaration = _document(tmp_path / "declaration.tcl", declaration_source)
usage = _document(tmp_path / "usage.tcl", usage_source)
server = TclLanguageServer(name="navigation-test", version="1", max_workers=1)
assert server.update_poco_completion_for_file(declaration)
assert server.update_poco_completion_for_file(usage)
monkeypatch.setattr(lsp_server, "LSP_SERVER", server)
position = _position(usage_source, "customProc")
identifier = lsp.TextDocumentIdentifier(uri=usage.uri)
definitions = lsp_server.goto_definition(
lsp.DefinitionParams(text_document=identifier, position=position)
)
assert definitions is not None
assert len(definitions) == 1
assert definitions[0].uri == declaration.uri
references = lsp_server.references(
lsp.ReferenceParams(
text_document=identifier,
position=position,
context=lsp.ReferenceContext(include_declaration=True),
)
)
assert len(references) == 2
prepared = lsp_server.prepare_rename(
lsp.PrepareRenameParams(text_document=identifier, position=position)
)
assert prepared is not None
assert prepared.placeholder == "customProc"
edit = lsp_server.rename(
lsp.RenameParams(
text_document=identifier,
position=position,
new_name="renamedProc",
)
)
assert edit is not None
assert edit.changes is not None
assert set(edit.changes) == {declaration.uri, usage.uri}
assert all(
text_edit.new_text == "renamedProc"
for edits in edit.changes.values()
for text_edit in edits
)
symbols = lsp_server.workspace_symbol(
lsp.WorkspaceSymbolParams(query="custom")
)
assert [symbol.name for symbol in symbols] == ["customProc"]
def test_duplicate_proc_definition_cannot_be_renamed(tmp_path: Path, monkeypatch):
server = TclLanguageServer(name="navigation-test", version="1", max_workers=1)
documents = [
_document(tmp_path / f"duplicate_{number}.tcl", "proc duplicate {} { return }")
for number in range(2)
]
for document in documents:
assert server.update_poco_completion_for_file(document)
monkeypatch.setattr(lsp_server, "LSP_SERVER", server)
result = lsp_server.prepare_rename(
lsp.PrepareRenameParams(
text_document=lsp.TextDocumentIdentifier(uri=documents[0].uri),
position=lsp.Position(line=0, character=6),
)
)
assert result is None