Compare commits

...
3 Commits
Author SHA1 Message Date
Christoph Brandau af195a577b refactor(lsp): cache analysis results and debounce diagnostics
build_and_puplish.yml / build_and_publish (release) Successful in 29s
This change adds cached and incremental analysis for the LSP
server to improve responsiveness. The client now debounces
diagnostic updates to avoid excessive recomputation. The
server introduces per-document line caches and various
caches for completions, inlay hints, and metadata to
support faster, incremental updates.

- Debounce diagnostics on text changes to reduce noise.
- Add caches for completions, inlay hints, and metadata.
- Introduce incremental analysis with per-document line caches.
2026-08-19 13:41:53 +02:00
Christoph Brandau ecb50be2b8 feat(inlay-hints): add configurable parameter name hints
Adds configurable inlay hints for TCL procedures and merges signatures from built-ins and workspace files. The feature supports parameterNames and suppressWhenArgumentMatchesName and respects an optional range filter and current-file priority.

- Introduces built-in and custom inlay hint builders
- Honors inlayHints parameterNames and suppression options
- Adds tests validating hints, ranges, and priority rules
2026-08-19 12:59:51 +02:00
Christoph 61d4785775 Update version to 2026.8.100 2026-08-19 07:47:31 +00:00
14 changed files with 1098 additions and 181 deletions
+5
View File
@@ -1,8 +1,13 @@
## Unreleased ## Unreleased
- Prevent truncated TCL inlay hints and add configurable parameter hint modes
- Add inlay hints for built-in NX procedures, variadic arguments, and visible ranges
- Add inlay hint documentation and navigation to custom procedure definitions
- Add signature help for custom TCL procedures and built-in NX/MOM procedures - Add signature help for custom TCL procedures and built-in NX/MOM procedures
- Clean stale TCL indexes on close, delete, and rename operations - Clean stale TCL indexes on close, delete, and rename operations
- Make background parsing and index updates thread-safe - Make background parsing and index updates thread-safe
- Improve TCL response times with debounced edits and cached semantic, inlay, hover, completion, and variable indexes
- Debounce CDL/DEF diagnostics and remove per-line diagnostic logging
## [0.0.1] ## [0.0.1]
+6
View File
@@ -31,6 +31,12 @@ The extension can be configured through VS Code settings:
- `nx-post-support.interpreter` - Specify custom Python interpreter path for the language server - `nx-post-support.interpreter` - Specify custom Python interpreter path for the language server
- `nx-post-support.formatter` - Enable/disable the TCL formatter (default: false) - `nx-post-support.formatter` - Enable/disable the TCL formatter (default: false)
- `nx-post-support.inlayHint` - Enable/disable inlay Hints (default: true) - `nx-post-support.inlayHint` - Enable/disable inlay Hints (default: true)
- `nx-post-support.inlayHints.parameterNames` - Show parameter names for `all`, only `literals`, or `none` (default: `all`)
- `nx-post-support.inlayHints.suppressWhenArgumentMatchesName` - Hide redundant hints such as `value:` before `$value` (default: true)
TCL files default to unlimited inlay hint length so that VS Code does not
truncate later parameter names on a line. An explicit user setting for
`editor.inlayHints.maximumLength` still takes precedence.
## Usage ## Usage
+15 -8
View File
@@ -4,6 +4,8 @@ import {
createCdlEventHandlerSnippet createCdlEventHandlerSnippet
} from "./cdlEventHandler" } from "./cdlEventHandler"
const MACHINE_HEADER_REGEX = /^MACHINE\s+\S+/
export function formatCdlFile(content: string): string { export function formatCdlFile(content: string): string {
let indentLevel = 0 let indentLevel = 0
const formattedLines = [] const formattedLines = []
@@ -41,15 +43,19 @@ export function formatDefFile(content: string): string {
} }
export function isFirstLineMachine(content: string): boolean { export function isFirstLineMachine(content: string): boolean {
const lines = content.split("\n").map((line) => line.trim()) let lineStart = 0
for (const line of lines) { while (lineStart <= content.length) {
console.log(line) const newline = content.indexOf("\n", lineStart)
const lineEnd = newline === -1 ? content.length : newline
const line = content.slice(lineStart, lineEnd).trim()
if (line === "" || line.startsWith("#")) { if (line === "" || line.startsWith("#")) {
console.log("skipping line") if (newline === -1) {
return false
}
lineStart = newline + 1
continue continue
} }
const machineRegex = /^MACHINE\s+\S+/ return MACHINE_HEADER_REGEX.test(line)
return machineRegex.test(line)
} }
return false return false
} }
@@ -57,10 +63,11 @@ export function isFirstLineMachine(content: string): boolean {
export function diagnosticHandler(document: vscode.TextDocument) { export function diagnosticHandler(document: vscode.TextDocument) {
const diagnostics: vscode.Diagnostic[] = [] const diagnostics: vscode.Diagnostic[] = []
if (document.languageId === "cdl" || document.languageId === "def") { if (document.languageId === "cdl" || document.languageId === "def") {
if (!isFirstLineMachine(document.getText())) { const text = document.getText()
if (!isFirstLineMachine(text)) {
const range = new vscode.Range( const range = new vscode.Range(
document.positionAt(0), document.positionAt(0),
document.positionAt(document.getText().length) document.positionAt(text.length)
) )
const diagnostic = new vscode.Diagnostic( const diagnostic = new vscode.Diagnostic(
range, range,
+23 -3
View File
@@ -18,6 +18,12 @@ export interface ISettings {
interpreter: string[] interpreter: string[]
importStrategy: string importStrategy: string
showNotifications: string showNotifications: string
formatter: boolean
inlayHint: boolean
inlayHints: {
parameterNames: "all" | "literals" | "none"
suppressWhenArgumentMatchesName: boolean
}
} }
export function getExtensionSettings( export function getExtensionSettings(
@@ -80,7 +86,13 @@ export async function getWorkspaceSettings(
importStrategy: config.get<string>(`importStrategy`) ?? "useBundled", importStrategy: config.get<string>(`importStrategy`) ?? "useBundled",
showNotifications: config.get<string>(`showNotifications`) ?? "off", showNotifications: config.get<string>(`showNotifications`) ?? "off",
formatter: config.get<boolean>(`formatter`) ?? true, formatter: config.get<boolean>(`formatter`) ?? true,
inlayHint: config.get<boolean>(`inlayHint`) ?? true inlayHint: config.get<boolean>(`inlayHint`) ?? true,
inlayHints: {
parameterNames:
config.get<"all" | "literals" | "none">(`inlayHints.parameterNames`) ?? "all",
suppressWhenArgumentMatchesName:
config.get<boolean>(`inlayHints.suppressWhenArgumentMatchesName`) ?? true
}
} }
return workspaceSetting return workspaceSetting
} }
@@ -113,7 +125,13 @@ export async function getGlobalSettings(
importStrategy: getGlobalValue<string>(config, "importStrategy", "useBundled"), importStrategy: getGlobalValue<string>(config, "importStrategy", "useBundled"),
showNotifications: getGlobalValue<string>(config, "showNotifications", "off"), showNotifications: getGlobalValue<string>(config, "showNotifications", "off"),
formatter: config.get<boolean>(`formatter`) ?? true, formatter: config.get<boolean>(`formatter`) ?? true,
inlayHint: config.get<boolean>(`inlayHint`) ?? true inlayHint: config.get<boolean>(`inlayHint`) ?? true,
inlayHints: {
parameterNames:
config.get<"all" | "literals" | "none">(`inlayHints.parameterNames`) ?? "all",
suppressWhenArgumentMatchesName:
config.get<boolean>(`inlayHints.suppressWhenArgumentMatchesName`) ?? true
}
} }
return setting return setting
} }
@@ -129,7 +147,9 @@ export function checkIfConfigurationChanged(
`${namespace}.importStrategy`, `${namespace}.importStrategy`,
`${namespace}.showNotifications`, `${namespace}.showNotifications`,
`${namespace}.formatter`, `${namespace}.formatter`,
`${namespace}.inlayHint` `${namespace}.inlayHint`,
`${namespace}.inlayHints.parameterNames`,
`${namespace}.inlayHints.suppressWhenArgumentMatchesName`
] ]
const changed = settings.map((s) => e.affectsConfiguration(s)) const changed = settings.map((s) => e.affectsConfiguration(s))
return changed.includes(true) return changed.includes(true)
+50 -11
View File
@@ -246,28 +246,67 @@ export async function activate(context: vscode.ExtensionContext) {
const diagnosticCollectionDef = vscode.languages.createDiagnosticCollection("def") const diagnosticCollectionDef = vscode.languages.createDiagnosticCollection("def")
context.subscriptions.push(diagnosticCollectionCdl, diagnosticCollectionDef) context.subscriptions.push(diagnosticCollectionCdl, diagnosticCollectionDef)
const diagnosticTimers = new Map<string, ReturnType<typeof setTimeout>>()
const updateDiagnostics = (document: vscode.TextDocument) => {
if (document.languageId === "cdl") {
diagnosticCollectionCdl.set(document.uri, diagnosticHandler(document))
} else if (document.languageId === "def") {
diagnosticCollectionDef.set(document.uri, diagnosticHandler(document))
}
}
const scheduleDiagnostics = (document: vscode.TextDocument) => {
const key = document.uri.toString()
const previous = diagnosticTimers.get(key)
if (previous !== undefined) {
clearTimeout(previous)
}
diagnosticTimers.set(
key,
setTimeout(() => {
diagnosticTimers.delete(key)
updateDiagnostics(document)
}, 120)
)
}
// Check if the first line of the CDL file contains "MACHINE" // Check if the first line of the CDL file contains "MACHINE"
context.subscriptions.push( context.subscriptions.push(
vscode.workspace.onDidOpenTextDocument((document) => { vscode.workspace.onDidOpenTextDocument((document) => {
if (document.languageId === "cdl" || document.languageId === "def") { if (document.languageId === "cdl" || document.languageId === "def") {
if (document.languageId === "cdl") { updateDiagnostics(document)
diagnosticCollectionCdl.set(document.uri, diagnosticHandler(document))
} else if (document.languageId === "def") {
diagnosticCollectionDef.set(document.uri, diagnosticHandler(document))
}
} }
}), }),
vscode.workspace.onDidChangeTextDocument((event) => { vscode.workspace.onDidChangeTextDocument((event) => {
const document = event.document const document = event.document
if (document.languageId === "cdl" || document.languageId === "def") { if (document.languageId === "cdl" || document.languageId === "def") {
if (document.languageId === "cdl") { scheduleDiagnostics(document)
diagnosticCollectionCdl.set(document.uri, diagnosticHandler(document))
} else if (document.languageId === "def") {
diagnosticCollectionDef.set(document.uri, diagnosticHandler(document))
}
} }
}) }),
vscode.workspace.onDidCloseTextDocument((document) => {
const key = document.uri.toString()
const timer = diagnosticTimers.get(key)
if (timer !== undefined) {
clearTimeout(timer)
diagnosticTimers.delete(key)
}
diagnosticCollectionCdl.delete(document.uri)
diagnosticCollectionDef.delete(document.uri)
}),
{
dispose() {
for (const timer of diagnosticTimers.values()) {
clearTimeout(timer)
}
diagnosticTimers.clear()
}
}
) )
for (const document of vscode.workspace.textDocuments) {
if (document.languageId === "cdl" || document.languageId === "def") {
updateDiagnostics(document)
}
}
} }
export function deactivate(): Thenable<void> | undefined { export function deactivate(): Thenable<void> | undefined {
+26 -1
View File
@@ -2,7 +2,7 @@
"name": "nx-post-support", "name": "nx-post-support",
"displayName": "NX Postprocessor Support", "displayName": "NX Postprocessor Support",
"description": "VS Code extension for NX CAM postprocessor development with syntax highlighting, formatting, linting, and auto-completion for CDL, TCL, and DEF files", "description": "VS Code extension for NX CAM postprocessor development with syntax highlighting, formatting, linting, and auto-completion for CDL, TCL, and DEF files",
"version": "2026.6.201", "version": "2026.8.100",
"publisher": "Christoph", "publisher": "Christoph",
"icon": "images/nx-1.png", "icon": "images/nx-1.png",
"extensionDependencies": [ "extensionDependencies": [
@@ -96,6 +96,26 @@
"default": true, "default": true,
"description": "Use the Inlay Hints in from `NX Postprocessor Support`" "description": "Use the Inlay Hints in from `NX Postprocessor Support`"
}, },
"nx-post-support.inlayHints.parameterNames": {
"type": "string",
"default": "all",
"enum": [
"all",
"literals",
"none"
],
"enumDescriptions": [
"Show parameter name hints for all arguments.",
"Show parameter name hints only for literal arguments.",
"Do not show parameter name hints."
],
"description": "Controls which TCL procedure arguments receive parameter name hints."
},
"nx-post-support.inlayHints.suppressWhenArgumentMatchesName": {
"type": "boolean",
"default": true,
"description": "Hide a parameter hint when a variable argument already has the same name, for example `output` in `my_proc $output`."
},
"nx-post-support.importStrategy": { "nx-post-support.importStrategy": {
"default": "useBundled", "default": "useBundled",
"description": "Defines where `NX Postprocessor Support` is imported from.", "description": "Defines where `NX Postprocessor Support` is imported from.",
@@ -120,6 +140,11 @@
"type": "array" "type": "array"
} }
} }
},
"configurationDefaults": {
"[tcl]": {
"editor.inlayHints.maximumLength": 0
}
} }
}, },
"scripts": { "scripts": {
+96 -79
View File
@@ -5,15 +5,15 @@
from __future__ import annotations from __future__ import annotations
import json import json
import operator
import os import os
import pathlib import pathlib
import re import re
import sys import sys
import threading import threading
from collections import ChainMap
from typing import Any, Optional
import operator
from functools import reduce from functools import reduce
from typing import Any, Optional
# ********************************************************** # **********************************************************
@@ -40,11 +40,15 @@ update_sys_path(
# pylint: disable=wrong-import-position,import-error # pylint: disable=wrong-import-position,import-error
import lsp_jsonrpc as jsonrpc import lsp_jsonrpc as jsonrpc
import lsprotocol.types as lsp import lsprotocol.types as lsp
from pygls import uris, workspace
from common.load_data import standard_items from common.load_data import standard_items
from lsp_tclserver import TclLanguageServer
from pygls import uris, workspace
from pygls.workspace.text_document import TextDocument
from tools.folding_ranges import build_folding_ranges from tools.folding_ranges import build_folding_ranges
from tools.semantic_tokens import _Highlighter, TOKEN_TYPES, TokenModifier from tools.inlay_hint import (
from tools.inlay_hint import InlayHintGenerator InlayHintGenerator,
build_builtin_inlay_signatures,
)
from tools.navigation import ( from tools.navigation import (
SymbolIdentity, SymbolIdentity,
definition_identities, definition_identities,
@@ -52,10 +56,13 @@ from tools.navigation import (
symbol_at_position, symbol_at_position,
workspace_symbols, workspace_symbols,
) )
from tools.semantic_tokens import (
TOKEN_TYPE_INDEX,
TOKEN_TYPES,
TokenModifier,
_Highlighter,
)
from tools.signature_help import build_signature_help from tools.signature_help import build_signature_help
from lsp_tclserver import TclLanguageServer
from pygls.workspace.text_document import TextDocument
WORKSPACE_SETTINGS = {} WORKSPACE_SETTINGS = {}
GLOBAL_SETTINGS = {} GLOBAL_SETTINGS = {}
@@ -71,6 +78,23 @@ BUILTIN_PROC_NAMES = {
for item in standard_items.tcl_keyword_list + standard_items.nx_procs for item in standard_items.tcl_keyword_list + standard_items.nx_procs
} }
BUILTIN_VARIABLE_NAMES = {item.label for item in standard_items.nx_variables} BUILTIN_VARIABLE_NAMES = {item.label for item in standard_items.nx_variables}
STATIC_COMPLETION_ITEMS = tuple(
standard_items.tcl_keyword_list
+ standard_items.nx_procs
+ standard_items.nx_variables
)
STATIC_COMPLETION_KEYS = frozenset(
(item.label, getattr(item, "kind", None)) for item in STATIC_COMPLETION_ITEMS
)
BUILTIN_INLAY_SIGNATURES = build_builtin_inlay_signatures(
standard_items.json_data.get("MOM_procs", [])
)
BUILTIN_HOVER_ITEMS = {}
for _hover_item in (
standard_items.json_data.get("MOM_procs", [])
+ standard_items.json_data.get("mom_variables", [])
):
BUILTIN_HOVER_ITEMS.setdefault(_hover_item.get("label"), _hover_item)
# ********************************************************** # **********************************************************
# Tool specific code goes below this. # Tool specific code goes below this.
@@ -91,15 +115,14 @@ def did_open(params: lsp.DidOpenTextDocumentParams) -> None:
"""LSP handler for textDocument/didOpen request.""" """LSP handler for textDocument/didOpen request."""
document = LSP_SERVER.workspace.get_text_document(params.text_document.uri) document = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
LSP_SERVER.clear_cache_for_uri(document.uri) LSP_SERVER.clear_cache_for_uri(document.uri)
LSP_SERVER.compute_diagnostics(document) LSP_SERVER.analyze_document_now(document)
# Also update custom completion and proc docs for this file
LSP_SERVER.update_poco_completion_for_file(document)
@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DID_SAVE) @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DID_SAVE)
def did_save(params: lsp.DidSaveTextDocumentParams) -> None: def did_save(params: lsp.DidSaveTextDocumentParams) -> None:
"""LSP handler for textDocument/didSave request.""" """LSP handler for textDocument/didSave request."""
_ = LSP_SERVER.workspace.get_text_document(params.text_document.uri) document = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
LSP_SERVER.analyze_document_now(document)
@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DID_CLOSE) @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DID_CLOSE)
@@ -115,8 +138,7 @@ def did_change(params: lsp.DidChangeTextDocumentParams) -> None:
"""LSP handler for textDocument/didChange request""" """LSP handler for textDocument/didChange request"""
document = LSP_SERVER.workspace.get_text_document(params.text_document.uri) document = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
LSP_SERVER.clear_cache_for_uri(document.uri) LSP_SERVER.clear_cache_for_uri(document.uri)
LSP_SERVER.compute_diagnostics(document) LSP_SERVER.schedule_document_analysis(document)
LSP_SERVER.update_poco_completion_for_file(document)
FILE_OPERATION_OPTIONS = lsp.FileOperationRegistrationOptions( FILE_OPERATION_OPTIONS = lsp.FileOperationRegistrationOptions(
@@ -199,10 +221,12 @@ def did_change_watched_files(params: lsp.DidChangeWatchedFilesParams) -> None:
def document_diagnostic(params: lsp.DocumentDiagnosticParams): def document_diagnostic(params: lsp.DocumentDiagnosticParams):
"""Return diagnostics for the requested document""" """Return diagnostics for the requested document"""
uri = params.text_document.uri uri = params.text_document.uri
doc = LSP_SERVER.workspace.get_text_document(uri)
diagnostic_state = LSP_SERVER.diagnostic_snapshot(uri) diagnostic_state = LSP_SERVER.diagnostic_snapshot(uri)
was_cached = diagnostic_state is not None was_cached = (
if diagnostic_state is None: diagnostic_state is not None and diagnostic_state[0] == doc.version
doc = LSP_SERVER.workspace.get_text_document(uri) )
if not was_cached:
LSP_SERVER.compute_diagnostics(doc) LSP_SERVER.compute_diagnostics(doc)
diagnostic_state = LSP_SERVER.diagnostic_snapshot(uri) diagnostic_state = LSP_SERVER.diagnostic_snapshot(uri)
@@ -220,24 +244,15 @@ def document_diagnostic(params: lsp.DocumentDiagnosticParams):
@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_COMPLETION) @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_COMPLETION)
def on_completion(params: lsp.CompletionParams) -> lsp.CompletionList: def on_completion(params: lsp.CompletionParams) -> lsp.CompletionList:
from tools.variable_index import build_variable_index
from tools.completion_items import BUILTIN_VAR_LABELS from tools.completion_items import BUILTIN_VAR_LABELS
doc = LSP_SERVER.workspace.get_text_document(params.text_document.uri) doc = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
# Base items workspace_items = LSP_SERVER.completion_items_snapshot()
poco_completion, _, _ = LSP_SERVER.index_snapshot()
poco = [item for items in poco_completion.values() for item in items]
base_items = (
standard_items.tcl_keyword_list
+ standard_items.nx_procs
+ standard_items.nx_variables
+ poco
)
# Build variable index from current document
tree = LSP_SERVER.get_tree(doc) tree = LSP_SERVER.get_tree(doc)
globals_set, procs_locals, proc_ranges = build_variable_index(doc.source, tree) globals_set, procs_locals, proc_ranges = LSP_SERVER.variable_index_for_document(
doc, tree
)
# Always include globals (excluding built-ins) # Always include globals (excluding built-ins)
dynamic_items = [] dynamic_items = []
@@ -264,9 +279,11 @@ def on_completion(params: lsp.CompletionParams) -> lsp.CompletionList:
# Merge with de-duplication. Each file keeps its complete index, so a proc # Merge with de-duplication. Each file keeps its complete index, so a proc
# declared in multiple files must only appear once in the completion list. # declared in multiple files must only appear once in the completion list.
merged: list[lsp.CompletionItem] = [] merged: list[lsp.CompletionItem] = list(STATIC_COMPLETION_ITEMS)
seen_items: set[tuple[str, lsp.CompletionItemKind | None]] = set() seen_items: set[tuple[str, lsp.CompletionItemKind | None]] = set(
for it in base_items + dynamic_items: STATIC_COMPLETION_KEYS
)
for it in (*workspace_items, *dynamic_items):
key = (it.label, getattr(it, "kind", None)) key = (it.label, getattr(it, "kind", None))
if key in seen_items: if key in seen_items:
continue continue
@@ -287,22 +304,9 @@ def signature_help(params: lsp.SignatureHelpParams) -> lsp.SignatureHelp | None:
document = LSP_SERVER.workspace.get_text_document(params.text_document.uri) document = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
tree = LSP_SERVER.get_tree(document) tree = LSP_SERVER.get_tree(document)
filepath = str(pathlib.Path(uris.to_fs_path(document.uri))) custom_signatures, custom_docs = LSP_SERVER.proc_metadata_snapshot(
custom_signatures: dict[str, list[str]] = {} document.path
custom_docs: dict[str, str] = {} )
_, proc_signatures, proc_docs = LSP_SERVER.index_snapshot()
# Prefer declarations from the current document if duplicate proc names
# exist in the workspace.
for indexed_path, signatures in proc_signatures.items():
if indexed_path != filepath:
custom_signatures.update(signatures)
custom_signatures.update(proc_signatures.get(filepath, {}))
for indexed_path, docs in proc_docs.items():
if indexed_path != filepath:
custom_docs.update(docs)
custom_docs.update(proc_docs.get(filepath, {}))
return build_signature_help( return build_signature_help(
document.source, document.source,
@@ -333,22 +337,37 @@ def document_symbols(params: lsp.DocumentSymbolParams):
@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_INLAY_HINT) @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_INLAY_HINT)
def inlay_hints(params: lsp.InlayHintParams): def inlay_hints(params: lsp.InlayHintParams):
if not GLOBAL_SETTINGS.get("inlayHint", False):
return []
document = LSP_SERVER.workspace.get_text_document(params.text_document.uri) document = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
settings = _get_settings_by_document(document)
if not settings.get("inlayHint", False):
return []
inlay_settings = settings.get("inlayHints", {})
parameter_names = inlay_settings.get("parameterNames", "all")
if parameter_names == "none":
return []
# Reuse cached AST # Reuse cached AST
tree = LSP_SERVER.get_tree(document) tree = LSP_SERVER.get_tree(document)
# Merge proc signatures across files and traverse once # Built-in NX procedures are the fallback. Workspace procedures replace them,
merged_signatures = {} # and a declaration in the current file wins over duplicate workspace names.
_, proc_signatures, _ = LSP_SERVER.index_snapshot() custom_signatures = LSP_SERVER.custom_inlay_signatures_snapshot(
for sigs in proc_signatures.values(): document.path
merged_signatures.update(sigs) )
signatures = ChainMap(custom_signatures, BUILTIN_INLAY_SIGNATURES)
generator = InlayHintGenerator(merged_signatures) generator = InlayHintGenerator(
tree.accept(generator, recurse=True) document.source,
return generator.hints signatures,
source_lines=LSP_SERVER.get_lines(document),
requested_range=params.range,
parameter_names=parameter_names,
suppress_when_argument_matches_name=inlay_settings.get(
"suppressWhenArgumentMatchesName", True
),
)
return generator.generate(tree)
@LSP_SERVER.feature( @LSP_SERVER.feature(
@@ -363,8 +382,7 @@ def semantic_tokens(params: lsp.SemanticTokensParams):
data = [] data = []
plugins = [] plugins = []
poco_completion, _, _ = LSP_SERVER.index_snapshot() hl = _Highlighter(plugins, LSP_SERVER.custom_function_names_snapshot())
hl = _Highlighter(plugins, poco_completion)
# Reuse cached AST # Reuse cached AST
tree = LSP_SERVER.get_tree(document) tree = LSP_SERVER.get_tree(document)
@@ -377,7 +395,7 @@ def semantic_tokens(params: lsp.SemanticTokensParams):
token.line, token.line,
token.offset, token.offset,
token.length, token.length,
TOKEN_TYPES.index(token.tok_type), TOKEN_TYPE_INDEX[token.tok_type],
reduce(operator.or_, token.tok_modifiers, 0), reduce(operator.or_, token.tok_modifiers, 0),
] ]
) )
@@ -399,14 +417,14 @@ def hover(params: lsp.HoverParams) -> lsp.Hover:
col = params.position.character col = params.position.character
try: try:
line = document.lines[pos.line] line = LSP_SERVER.get_lines(document)[pos.line]
except IndexError: except IndexError:
return None return None
# Do not show hover for proc name in its declaration # Do not show hover for proc name in its declaration
from tools.proc_docs import is_proc_declaration_position from tools.proc_docs import is_proc_declaration_line
if is_proc_declaration_position(document.source, pos.line, pos.character): if is_proc_declaration_line(line, pos.character):
return None return None
# Identify the token under the cursor # Identify the token under the cursor
@@ -418,11 +436,7 @@ def hover(params: lsp.HoverParams) -> lsp.Hover:
return None return None
# 1) If token is a known MOM proc/variable, return built-in hover # 1) If token is a known MOM proc/variable, return built-in hover
command = token match = BUILTIN_HOVER_ITEMS.get(token)
data = standard_items.json_data
all_items = data.get("MOM_procs", []) + data.get("mom_variables", [])
match = next((item for item in all_items if item["label"] == command), None)
if match and match.get("kind") == "function": if match and match.get("kind") == "function":
label = match.get("label", "") label = match.get("label", "")
parameters = match.get("parameters", []) parameters = match.get("parameters", [])
@@ -455,14 +469,10 @@ def hover(params: lsp.HoverParams) -> lsp.Hover:
# 2) Otherwise, check if the token is a custom proc and show its preceding doc block # 2) Otherwise, check if the token is a custom proc and show its preceding doc block
# Build a merged map of proc -> docs gathered during initialization and updates # Build a merged map of proc -> docs gathered during initialization and updates
proc_docs: dict[str, str] = {} proc_doc = LSP_SERVER.proc_documentation(token, document.path)
_, _, indexed_proc_docs = LSP_SERVER.index_snapshot() if proc_doc is not None:
for file_docs in indexed_proc_docs.values():
proc_docs.update(file_docs)
if token in proc_docs:
return lsp.Hover( return lsp.Hover(
lsp.MarkupContent(kind=lsp.MarkupKind.Markdown, value=proc_docs[token]) lsp.MarkupContent(kind=lsp.MarkupKind.Markdown, value=proc_doc)
) )
return None return None
@@ -490,7 +500,7 @@ def _navigation_context(uri: str, position: lsp.Position):
indexes = LSP_SERVER.navigation_snapshot() indexes = LSP_SERVER.navigation_snapshot()
filepath = str(pathlib.Path(uris.to_fs_path(uri))) filepath = str(pathlib.Path(uris.to_fs_path(uri)))
index = indexes.get(filepath) index = indexes.get(filepath)
if index is None: if index is None or LSP_SERVER.index_update_pending(filepath):
document = LSP_SERVER.workspace.get_text_document(uri) document = LSP_SERVER.workspace.get_text_document(uri)
LSP_SERVER.update_poco_completion_for_file(document) LSP_SERVER.update_poco_completion_for_file(document)
indexes = LSP_SERVER.navigation_snapshot() indexes = LSP_SERVER.navigation_snapshot()
@@ -769,6 +779,13 @@ def _get_global_defaults():
"showNotifications": GLOBAL_SETTINGS.get("showNotifications", "off"), "showNotifications": GLOBAL_SETTINGS.get("showNotifications", "off"),
"formatter": GLOBAL_SETTINGS.get("formatter", True), "formatter": GLOBAL_SETTINGS.get("formatter", True),
"inlayHint": GLOBAL_SETTINGS.get("inlayHint", True), "inlayHint": GLOBAL_SETTINGS.get("inlayHint", True),
"inlayHints": GLOBAL_SETTINGS.get(
"inlayHints",
{
"parameterNames": "all",
"suppressWhenArgumentMatchesName": True,
},
),
} }
+320 -5
View File
@@ -11,15 +11,16 @@ from pygls.workspace.text_document import TextDocument
from tclint.format import FormatterOpts from tclint.format import FormatterOpts
from tclint.lexer import TclSyntaxError from tclint.lexer import TclSyntaxError
from tclint.violations import Violation from tclint.violations import Violation
from tools import checks, parser from tools import checks, parser
from tools.completion_items import CompletionCollector from tools.completion_items import CompletionCollector
from tools.formatter import NxFormatter as Formatter from tools.formatter import NxFormatter as Formatter
from tools.inlay_hint import InlayHintSignature, build_custom_inlay_signatures
from tools.navigation import FileSymbolIndex, build_file_symbol_index from tools.navigation import FileSymbolIndex, build_file_symbol_index
from tools.proc_docs import build_proc_docs from tools.proc_docs import build_proc_docs
from tools.variable_index import ProcRange, build_variable_index
DIAGNOSTIC_SOURCE = "nx-post-support" DIAGNOSTIC_SOURCE = "nx-post-support"
LOGGER = logging.getLogger(__name__)
class TclLanguageServer(server.LanguageServer): class TclLanguageServer(server.LanguageServer):
@@ -33,15 +34,40 @@ class TclLanguageServer(server.LanguageServer):
self.proc_signatures: dict = {} self.proc_signatures: dict = {}
self.proc_docs: dict = {} self.proc_docs: dict = {}
self.navigation_indexes: dict[str, FileSymbolIndex] = {} self.navigation_indexes: dict[str, FileSymbolIndex] = {}
self.variable_indexes: dict[
str,
tuple[
int | None,
tuple[set[str], dict[str, set[str]], list[ProcRange]],
],
] = {}
# Cache: (uri, version) -> (tree, violations) # Cache: (uri, version) -> (tree, violations)
self._ast_cache = {} self._ast_cache = {}
self._line_cache: dict[tuple[str, int | None], tuple[str, ...]] = {}
self._parser_lock = threading.RLock() self._parser_lock = threading.RLock()
self._index_lock = threading.RLock() self._index_lock = threading.RLock()
self._index_tokens: dict[str, int] = {} self._index_tokens: dict[str, int] = {}
self._index_versions: dict[str, int | None] = {} self._index_versions: dict[str, int | None] = {}
self._committed_index_versions: dict[str, int | None] = {}
self._next_index_token = 0 self._next_index_token = 0
self._diagnostic_tokens: dict[str, int] = {} self._diagnostic_tokens: dict[str, int] = {}
self._next_diagnostic_token = 0 self._next_diagnostic_token = 0
self._index_generation = 0
self._workspace_completion_cache: tuple[int, tuple] = (-1, ())
self._custom_function_names_cache: tuple[int, frozenset[str]] = (
-1,
frozenset(),
)
self._proc_metadata_cache: dict[
str, tuple[int, dict[str, list[str]], dict[str, str]]
] = {}
self._custom_inlay_cache: dict[
str, tuple[int, dict[str, InlayHintSignature]]
] = {}
self._analysis_lock = threading.RLock()
self._analysis_timers: dict[str, threading.Timer] = {}
self._analysis_tokens: dict[str, int] = {}
self._next_analysis_token = 0
def _parse_source(self, source: str): def _parse_source(self, source: str):
self.parser.violations = [] self.parser.violations = []
@@ -74,11 +100,24 @@ class TclLanguageServer(server.LanguageServer):
self._ast_cache[key] = (tree, violations) self._ast_cache[key] = (tree, violations)
return tree, violations return tree, violations
def get_lines(self, document: TextDocument) -> tuple[str, ...]:
"""Return split source lines once per document version."""
key = (document.uri, document.version)
with self._parser_lock:
lines = self._line_cache.get(key)
if lines is None:
lines = tuple(document.source.splitlines())
self._line_cache[key] = lines
return lines
def clear_cache_for_uri(self, uri: str): def clear_cache_for_uri(self, uri: str):
with self._parser_lock: with self._parser_lock:
to_delete = [key for key in self._ast_cache if key[0] == uri] to_delete = [key for key in self._ast_cache if key[0] == uri]
for key in to_delete: for key in to_delete:
del self._ast_cache[key] del self._ast_cache[key]
for key in list(self._line_cache):
if key[0] == uri:
del self._line_cache[key]
@staticmethod @staticmethod
def _normalized_path(path: pathlib.Path | str) -> str: def _normalized_path(path: pathlib.Path | str) -> str:
@@ -101,6 +140,263 @@ class TclLanguageServer(server.LanguageServer):
) -> bool: ) -> bool:
return cls._normalized_path(first) == cls._normalized_path(second) return cls._normalized_path(first) == cls._normalized_path(second)
def _invalidate_workspace_caches_locked(self) -> None:
"""Invalidate request-level aggregates after an index mutation."""
self._index_generation += 1
self._workspace_completion_cache = (-1, ())
self._custom_function_names_cache = (-1, frozenset())
self._proc_metadata_cache.clear()
self._custom_inlay_cache.clear()
def completion_items_snapshot(self) -> tuple:
"""Return de-duplicated workspace completion items, cached by generation."""
with self._index_lock:
generation, items = self._workspace_completion_cache
if generation == self._index_generation:
return items
merged = []
seen = set()
for path_items in self.poco_completion.values():
for item in path_items:
key = (item.label, getattr(item, "kind", None))
if key in seen:
continue
seen.add(key)
merged.append(item)
items = tuple(merged)
self._workspace_completion_cache = (self._index_generation, items)
return items
def custom_function_names_snapshot(self) -> frozenset[str]:
"""Return custom completion labels for semantic highlighting."""
with self._index_lock:
generation, names = self._custom_function_names_cache
if generation == self._index_generation:
return names
names = frozenset(
item.label
for path_items in self.poco_completion.values()
for item in path_items
)
self._custom_function_names_cache = (self._index_generation, names)
return names
def proc_metadata_snapshot(
self, current_path: pathlib.Path | str
) -> tuple[dict[str, list[str]], dict[str, str]]:
"""Return merged proc metadata, preferring declarations in the active file."""
normalized_current = self._normalized_path(current_path)
with self._index_lock:
cached = self._proc_metadata_cache.get(normalized_current)
if cached is not None and cached[0] == self._index_generation:
return cached[1], cached[2]
signatures: dict[str, list[str]] = {}
docs: dict[str, str] = {}
signature_paths = sorted(
self.proc_signatures,
key=lambda path: self._normalized_path(path).casefold(),
)
doc_paths = sorted(
self.proc_docs,
key=lambda path: self._normalized_path(path).casefold(),
)
for path in signature_paths:
if self._normalized_path(path) != normalized_current:
signatures.update(self.proc_signatures[path])
for path in signature_paths:
if self._normalized_path(path) == normalized_current:
signatures.update(self.proc_signatures[path])
for path in doc_paths:
if self._normalized_path(path) != normalized_current:
docs.update(self.proc_docs[path])
for path in doc_paths:
if self._normalized_path(path) == normalized_current:
docs.update(self.proc_docs[path])
cached_value = (self._index_generation, signatures, docs)
self._proc_metadata_cache[normalized_current] = cached_value
return signatures, docs
def proc_documentation(
self, name: str, current_path: pathlib.Path | str
) -> str | None:
_, docs = self.proc_metadata_snapshot(current_path)
return docs.get(name)
def custom_inlay_signatures_snapshot(
self, current_path: pathlib.Path | str
) -> dict[str, InlayHintSignature]:
"""Return custom inlay signatures cached until the workspace index changes."""
normalized_current = self._normalized_path(current_path)
with self._index_lock:
cached = self._custom_inlay_cache.get(normalized_current)
if cached is not None and cached[0] == self._index_generation:
return cached[1]
signatures = build_custom_inlay_signatures(
self.proc_signatures,
self.proc_docs,
self.navigation_indexes,
os.fspath(current_path),
)
self._custom_inlay_cache[normalized_current] = (
self._index_generation,
signatures,
)
return signatures
def variable_index_for_document(
self, document: TextDocument, tree=None
) -> tuple[set[str], dict[str, set[str]], list[ProcRange]]:
"""Return the per-version variable index used by completion requests."""
filepath = str(pathlib.Path(uris.to_fs_path(document.uri)))
with self._index_lock:
cached = self.variable_indexes.get(filepath)
if cached is not None and cached[0] == document.version:
return cached[1]
if tree is None:
tree = self.get_tree(document)
variable_index = build_variable_index(document.source, tree)
with self._index_lock:
cached = self.variable_indexes.get(filepath)
if (
cached is None
or cached[0] is None
or document.version is None
or cached[0] <= document.version
):
self.variable_indexes[filepath] = (document.version, variable_index)
return variable_index
return cached[1]
def index_is_current(self, document: TextDocument) -> bool:
filepath = str(pathlib.Path(uris.to_fs_path(document.uri)))
with self._index_lock:
return (
filepath in self._committed_index_versions
and self._committed_index_versions[filepath] == document.version
)
def index_update_pending(self, filepath: pathlib.Path | str) -> bool:
filepath = os.fspath(filepath)
with self._index_lock:
return (
filepath not in self._committed_index_versions
or self._index_versions.get(filepath)
!= self._committed_index_versions[filepath]
)
def _cancel_document_analysis(self, uri: str) -> None:
with self._analysis_lock:
timer = self._analysis_timers.pop(uri, None)
self._analysis_tokens.pop(uri, None)
if timer is not None:
timer.cancel()
def cancel_analysis_under_uri(self, uri: str) -> None:
"""Cancel delayed analysis for a closed/deleted file or folder."""
try:
target = pathlib.Path(uris.to_fs_path(uri))
except (TypeError, ValueError):
self._cancel_document_analysis(uri)
return
with self._analysis_lock:
matching_uris = []
for pending_uri in self._analysis_timers:
try:
pending_path = pathlib.Path(uris.to_fs_path(pending_uri))
except (TypeError, ValueError):
continue
if self._is_same_or_child(pending_path, target):
matching_uris.append(pending_uri)
timers = [self._analysis_timers.pop(key) for key in matching_uris]
for key in matching_uris:
self._analysis_tokens.pop(key, None)
for timer in timers:
timer.cancel()
def _invalidate_document_work(self, document: TextDocument) -> None:
"""Prevent older diagnostic/index work from committing after a new edit."""
filepath = str(pathlib.Path(uris.to_fs_path(document.uri)))
with self._index_lock:
self._next_diagnostic_token += 1
self._diagnostic_tokens[document.uri] = self._next_diagnostic_token
self._next_index_token += 1
self._index_tokens[filepath] = self._next_index_token
self._index_versions[filepath] = document.version
def schedule_document_analysis(
self, document: TextDocument, delay_seconds: float = 0.15
) -> None:
"""Coalesce rapid edits and analyze only the latest immutable snapshot."""
snapshot = TextDocument(
uri=document.uri,
source=document.source,
version=document.version,
language_id=document.language_id,
)
self._invalidate_document_work(snapshot)
with self._analysis_lock:
previous = self._analysis_timers.pop(snapshot.uri, None)
if previous is not None:
previous.cancel()
self._next_analysis_token += 1
token = self._next_analysis_token
self._analysis_tokens[snapshot.uri] = token
def analyze() -> None:
with self._analysis_lock:
if self._analysis_tokens.get(snapshot.uri) != token:
return
try:
diagnostic_state = self.diagnostic_snapshot(snapshot.uri)
if (
diagnostic_state is None
or diagnostic_state[0] != snapshot.version
):
self.compute_diagnostics(snapshot)
with self._analysis_lock:
if self._analysis_tokens.get(snapshot.uri) != token:
return
if not self.index_is_current(snapshot):
self.update_poco_completion_for_file(snapshot)
except Exception:
LOGGER.exception("Delayed analysis failed for %s", snapshot.uri)
finally:
with self._analysis_lock:
if self._analysis_tokens.get(snapshot.uri) == token:
self._analysis_tokens.pop(snapshot.uri, None)
self._analysis_timers.pop(snapshot.uri, None)
timer = threading.Timer(delay_seconds, analyze)
timer.daemon = True
self._analysis_timers[snapshot.uri] = timer
timer.start()
def analyze_document_now(self, document: TextDocument) -> None:
"""Cancel delayed work and synchronously analyze the current document."""
self._cancel_document_analysis(document.uri)
self._invalidate_document_work(document)
diagnostic_state = self.diagnostic_snapshot(document.uri)
if diagnostic_state is None or diagnostic_state[0] != document.version:
self.compute_diagnostics(document)
if not self.index_is_current(document):
self.update_poco_completion_for_file(document)
def index_snapshot(self) -> tuple[dict, dict, dict]: def index_snapshot(self) -> tuple[dict, dict, dict]:
"""Return stable copies for request handlers running beside the indexer.""" """Return stable copies for request handlers running beside the indexer."""
with self._index_lock: with self._index_lock:
@@ -148,6 +444,9 @@ class TclLanguageServer(server.LanguageServer):
self.proc_signatures.pop(filepath, None) self.proc_signatures.pop(filepath, None)
self.proc_docs.pop(filepath, None) self.proc_docs.pop(filepath, None)
self.navigation_indexes.pop(filepath, None) self.navigation_indexes.pop(filepath, None)
self.variable_indexes.pop(filepath, None)
self._committed_index_versions.pop(filepath, None)
self._invalidate_workspace_caches_locked()
def indexed_paths_under_uri(self, uri: str) -> list[pathlib.Path]: def indexed_paths_under_uri(self, uri: str) -> list[pathlib.Path]:
target = pathlib.Path(uris.to_fs_path(uri)) target = pathlib.Path(uris.to_fs_path(uri))
@@ -156,6 +455,7 @@ class TclLanguageServer(server.LanguageServer):
indexed_paths.update(self.proc_signatures) indexed_paths.update(self.proc_signatures)
indexed_paths.update(self.proc_docs) indexed_paths.update(self.proc_docs)
indexed_paths.update(self.navigation_indexes) indexed_paths.update(self.navigation_indexes)
indexed_paths.update(self.variable_indexes)
indexed_paths.update(self._index_tokens) indexed_paths.update(self._index_tokens)
return [ return [
pathlib.Path(path) pathlib.Path(path)
@@ -165,20 +465,28 @@ class TclLanguageServer(server.LanguageServer):
def remove_file_state(self, uri: str) -> None: def remove_file_state(self, uri: str) -> None:
"""Remove cached and indexed state for a file or a complete folder.""" """Remove cached and indexed state for a file or a complete folder."""
self.cancel_analysis_under_uri(uri)
target = pathlib.Path(uris.to_fs_path(uri)) target = pathlib.Path(uris.to_fs_path(uri))
with self._index_lock: with self._index_lock:
index_changed = False
for store in ( for store in (
self.poco_completion, self.poco_completion,
self.proc_signatures, self.proc_signatures,
self.proc_docs, self.proc_docs,
self.navigation_indexes, self.navigation_indexes,
self.variable_indexes,
self._index_tokens, self._index_tokens,
self._index_versions, self._index_versions,
self._committed_index_versions,
): ):
for path in list(store): for path in list(store):
if self._is_same_or_child(path, target): if self._is_same_or_child(path, target):
del store[path] del store[path]
index_changed = True
if index_changed:
self._invalidate_workspace_caches_locked()
diagnostic_uris = set(self.diagnostics) diagnostic_uris = set(self.diagnostics)
diagnostic_uris.update(self._diagnostic_tokens) diagnostic_uris.update(self._diagnostic_tokens)
@@ -192,13 +500,16 @@ class TclLanguageServer(server.LanguageServer):
self._diagnostic_tokens.pop(diagnostic_uri, None) self._diagnostic_tokens.pop(diagnostic_uri, None)
with self._parser_lock: with self._parser_lock:
for key in list(self._ast_cache): cached_keys = set(self._ast_cache)
cached_keys.update(self._line_cache)
for key in cached_keys:
try: try:
cached_path = pathlib.Path(uris.to_fs_path(key[0])) cached_path = pathlib.Path(uris.to_fs_path(key[0]))
except (TypeError, ValueError): except (TypeError, ValueError):
continue continue
if self._is_same_or_child(cached_path, target): if self._is_same_or_child(cached_path, target):
del self._ast_cache[key] self._ast_cache.pop(key, None)
self._line_cache.pop(key, None)
def update_poco_completion_for_file( def update_poco_completion_for_file(
self, self,
@@ -225,8 +536,9 @@ class TclLanguageServer(server.LanguageServer):
navigation_index = build_file_symbol_index( navigation_index = build_file_symbol_index(
filepath, document.uri, tree filepath, document.uri, tree
) )
variable_index = build_variable_index(document.source, tree)
except Exception as e: except Exception as e:
logging.debug(f"Error parsing {filepath}: {e}") LOGGER.debug("Error parsing %s: %s", filepath, e)
self._discard_index_update(filepath, token) self._discard_index_update(filepath, token)
return False return False
@@ -241,6 +553,9 @@ class TclLanguageServer(server.LanguageServer):
self.proc_signatures[filepath] = dict(collector.proc_signatures) self.proc_signatures[filepath] = dict(collector.proc_signatures)
self.proc_docs[filepath] = docs self.proc_docs[filepath] = docs
self.navigation_indexes[filepath] = navigation_index self.navigation_indexes[filepath] = navigation_index
self.variable_indexes[filepath] = (document.version, variable_index)
self._committed_index_versions[filepath] = document.version
self._invalidate_workspace_caches_locked()
return True return True
def format( def format(
+9 -4
View File
@@ -1,8 +1,9 @@
from tclint.syntax_tree import Visitor, Command, BareWord, List
import lsprotocol.types as lsp import lsprotocol.types as lsp
from common.load_data import standard_items from common.load_data import standard_items
from tclint.syntax_tree import BareWord, Command, List, Visitor
BUILTIN_VAR_LABELS = {ci.label for ci in standard_items.nx_variables} BUILTIN_VAR_LABELS = {ci.label for ci in standard_items.nx_variables}
BUILTIN_PROC_LABELS = {ci.label for ci in standard_items.nx_procs}
class CompletionItems: class CompletionItems:
@@ -22,6 +23,7 @@ class CompletionCollector(Visitor):
def __init__(self): def __init__(self):
super().__init__() super().__init__()
self._custom_functions: list[lsp.CompletionItem] = [] self._custom_functions: list[lsp.CompletionItem] = []
self._custom_function_keys: set[tuple[str, lsp.CompletionItemKind | None]] = set()
self._proc_signatures = {} self._proc_signatures = {}
@property @property
@@ -34,8 +36,11 @@ class CompletionCollector(Visitor):
def _append_unique(self, item: lsp.CompletionItem): def _append_unique(self, item: lsp.CompletionItem):
# Avoid duplicate labels within the same file scan # Avoid duplicate labels within the same file scan
if not any(ci.label == item.label for ci in self._custom_functions): key = (item.label, item.kind)
self._custom_functions.append(item) if key in self._custom_function_keys:
return
self._custom_function_keys.add(key)
self._custom_functions.append(item)
def visit_command(self, command: Command): def visit_command(self, command: Command):
routine = command.routine routine = command.routine
@@ -46,7 +51,7 @@ class CompletionCollector(Visitor):
if not getattr(first_arg, "value", None): if not getattr(first_arg, "value", None):
return return
if any(item.label == first_arg.value for item in standard_items.nx_procs): if first_arg.value in BUILTIN_PROC_LABELS:
return return
# Record proc name as a completion item # Record proc name as a completion item
+252 -16
View File
@@ -1,29 +1,265 @@
from __future__ import annotations
import os
import re
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from typing import Any
import lsprotocol.types as lsp import lsprotocol.types as lsp
from tclint.syntax_tree import Visitor, Command from tclint.syntax_tree import Command, VarSub, Visitor
from tools.navigation import FileSymbolIndex
@dataclass(frozen=True)
class InlayHintParameter:
name: str
documentation: str | None = None
variadic: bool = False
@dataclass(frozen=True)
class InlayHintSignature:
parameters: tuple[InlayHintParameter, ...]
display_label: str
documentation: str | None = None
location: lsp.Location | None = None
def _normalized_path(path: str) -> str:
return os.path.normcase(os.path.abspath(path))
def _definition_locations(
index: FileSymbolIndex | None,
) -> dict[str, lsp.Location]:
if index is None:
return {}
locations = {}
for occurrence in index.occurrences:
if (
occurrence.is_definition
and occurrence.identity.kind == "proc"
):
locations[occurrence.placeholder] = lsp.Location(
uri=index.uri, range=occurrence.range
)
return locations
def build_custom_inlay_signatures(
signatures_by_path: dict[str, dict[str, list[str]]],
docs_by_path: dict[str, dict[str, str]],
indexes_by_path: dict[str, FileSymbolIndex],
current_path: str,
) -> dict[str, InlayHintSignature]:
"""Merge workspace signatures deterministically, preferring the current file."""
current_normalized = _normalized_path(current_path)
paths = sorted(
signatures_by_path, key=lambda path: _normalized_path(path).casefold()
)
paths.sort(key=lambda path: _normalized_path(path) == current_normalized)
result: dict[str, InlayHintSignature] = {}
for path in paths:
docs = docs_by_path.get(path, {})
definition_locations = _definition_locations(indexes_by_path.get(path))
for proc_name, parameter_names in signatures_by_path[path].items():
parameters = tuple(
InlayHintParameter(
name=parameter_name,
variadic=(
parameter_name == "args"
and parameter_index == len(parameter_names) - 1
),
)
for parameter_index, parameter_name in enumerate(parameter_names)
)
result[proc_name] = InlayHintSignature(
parameters=parameters,
display_label=" ".join([proc_name, *parameter_names]),
documentation=docs.get(proc_name),
location=definition_locations.get(
proc_name.removeprefix("::").rsplit("::", 1)[-1]
),
)
return result
def _is_builtin_variadic(item: dict[str, Any], parameter_name: str, index: int) -> bool:
parameters = item.get("parameters", [])
if index != len(parameters) - 1:
return False
if "..." in parameter_name or "" in parameter_name:
return True
format_label = item.get("format", "")
return (
f"<{parameter_name}>+" in format_label or f"[{parameter_name}]+" in format_label
)
def _builtin_parameter_label(parameter_name: str) -> str:
if "..." not in parameter_name and "" not in parameter_name:
return parameter_name.strip("<>[]")
first_name = parameter_name.split()[0].strip("<>[]")
return re.sub(r"(?:_?1)$", "", first_name) or first_name
def build_builtin_inlay_signatures(
items: list[dict[str, Any]],
) -> dict[str, InlayHintSignature]:
result = {}
for item in items:
proc_name = item.get("label")
if not proc_name:
continue
parameters = tuple(
InlayHintParameter(
name=_builtin_parameter_label(parameter.get("name", "")),
documentation=parameter.get("desc") or None,
variadic=_is_builtin_variadic(
item, parameter.get("name", ""), parameter_index
),
)
for parameter_index, parameter in enumerate(item.get("parameters", []))
if parameter.get("name")
)
result[proc_name] = InlayHintSignature(
parameters=parameters,
display_label=item.get("format") or proc_name,
documentation=item.get("description") or None,
)
return result
def _position_in_range(
position: lsp.Position, requested_range: lsp.Range | None
) -> bool:
if requested_range is None:
return True
value = (position.line, position.character)
start = (requested_range.start.line, requested_range.start.character)
end = (requested_range.end.line, requested_range.end.character)
return start <= value < end
class InlayHintGenerator(Visitor): class InlayHintGenerator(Visitor):
def __init__(self, proc_signatures): def __init__(
self,
source: str,
proc_signatures: Mapping[str, InlayHintSignature],
*,
source_lines: Sequence[str] | None = None,
requested_range: lsp.Range | None = None,
parameter_names: str = "all",
suppress_when_argument_matches_name: bool = True,
):
self.source_lines = (
source_lines if source_lines is not None else source.splitlines()
)
self.proc_signatures = proc_signatures self.proc_signatures = proc_signatures
self.hints = [] self.requested_range = requested_range
self.parameter_names = parameter_names
self.suppress_when_argument_matches_name = suppress_when_argument_matches_name
self.hints: list[lsp.InlayHint] = []
def _node_intersects_requested_range(self, node) -> bool:
if self.requested_range is None:
return True
start = getattr(node, "pos", None)
end = getattr(node, "end_pos", None)
if start is None or end is None:
return True
node_start_line = start[0] - 1
node_end_line = end[0] - 1
return not (
node_end_line < self.requested_range.start.line
or node_start_line > self.requested_range.end.line
)
def generate(self, tree) -> list[lsp.InlayHint]:
"""Walk only syntax-tree branches overlapping the requested editor range."""
self.hints.clear()
def walk(node) -> None:
if not self._node_intersects_requested_range(node):
return
if isinstance(node, Command):
self.visit_command(node)
for child in getattr(node, "children", []):
walk(child)
walk(tree)
return self.hints
def _position(self, line: int, column: int) -> lsp.Position:
line_index = line - 1
character_index = column - 1
if 0 <= line_index < len(self.source_lines):
prefix = self.source_lines[line_index][:character_index]
character_index = len(prefix.encode("utf-16-le")) // 2
return lsp.Position(line=line_index, character=character_index)
@staticmethod
def _parameter_for_argument(
signature: InlayHintSignature, argument_index: int
) -> InlayHintParameter | None:
if argument_index < len(signature.parameters):
return signature.parameters[argument_index]
if signature.parameters and signature.parameters[-1].variadic:
return signature.parameters[-1]
return None
def _should_show(self, argument, parameter: InlayHintParameter) -> bool:
if self.parameter_names == "none":
return False
if self.parameter_names == "literals" and isinstance(argument, VarSub):
return False
if not self.suppress_when_argument_matches_name or not isinstance(
argument, VarSub
):
return True
return getattr(argument, "value", None) != parameter.name
@staticmethod
def _tooltip(signature: InlayHintSignature) -> lsp.MarkupContent:
value = f"`{signature.display_label}`"
if signature.documentation:
value += f"\n\n{signature.documentation}"
return lsp.MarkupContent(kind=lsp.MarkupKind.Markdown, value=value)
def visit_command(self, command: Command): def visit_command(self, command: Command):
name = getattr(command.routine, "contents", None) name = getattr(command.routine, "contents", None)
if name not in self.proc_signatures: signature = self.proc_signatures.get(name)
if signature is None or self.parameter_names == "none":
return return
param_names = self.proc_signatures[name] for argument_index, argument in enumerate(command.args):
for idx, arg in enumerate(command.args): parameter = self._parameter_for_argument(signature, argument_index)
if idx >= len(param_names): if parameter is None:
break break
param_name = param_names[idx] if not argument.pos or not self._should_show(argument, parameter):
continue
if arg.pos: line, column = argument.pos
line, col = arg.pos position = self._position(line, column)
self.hints.append( if not _position_in_range(position, self.requested_range):
lsp.InlayHint( continue
position=lsp.Position(line=line - 1, character=col - 1),
label=f"{param_name}:", label = lsp.InlayHintLabelPart(
kind=lsp.InlayHintKind.Parameter, value=f"{parameter.name}:",
) tooltip=parameter.documentation,
location=signature.location,
)
self.hints.append(
lsp.InlayHint(
position=position,
label=[label],
kind=lsp.InlayHintKind.Parameter,
tooltip=self._tooltip(signature),
) )
)
+17 -41
View File
@@ -1,8 +1,9 @@
import re import re
from typing import Dict, List from typing import Dict, List
from tclint.syntax_tree import Visitor, Command from tclint.syntax_tree import Command, Visitor
from tools.parser import CustomParser
PROC_DECLARATION_RE = re.compile(r"^\s*proc\s+([^\s\{]+)")
def _strip_comment_prefix(line: str) -> str: def _strip_comment_prefix(line: str) -> str:
@@ -135,43 +136,18 @@ def build_proc_docs(tree, source_text: str) -> Dict[str, str]:
return extractor.docs return extractor.docs
def is_proc_declaration_position(source_text: str, line_zero_based: int, char_zero_based: int) -> bool: def is_proc_declaration_line(line: str, char_zero_based: int) -> bool:
"""Return True if the position is on the proc name on this source line."""
match = PROC_DECLARATION_RE.match(line)
return bool(match and match.start(1) <= char_zero_based <= match.end(1))
def is_proc_declaration_position(
source_text: str, line_zero_based: int, char_zero_based: int
) -> bool:
"""Return True if the position is on a proc name within its declaration.""" """Return True if the position is on a proc name within its declaration."""
parser = CustomParser() try:
tree = parser.parse(source_text) line = source_text.splitlines()[line_zero_based]
except IndexError:
# Walk commands to find 'proc' declarations and check if position intersects the name arg return False
class _DeclFinder(Visitor): return is_proc_declaration_line(line, char_zero_based)
def __init__(self):
self.is_decl = False
def visit_command(self, command: Command):
if self.is_decl:
return
routine = getattr(command.routine, "contents", None)
if routine != "proc" or not command.args:
return
name_node = command.args[0]
if not hasattr(name_node, "pos"):
return
# Calculate range for the name token
try:
start_line, start_col = name_node.pos
end_line, end_col = getattr(name_node, "end_pos", name_node.pos)
except Exception:
return
if start_line - 1 == line_zero_based:
length = 0
if hasattr(name_node, "value") and name_node.value is not None:
length = len(name_node.value)
elif hasattr(name_node, "contents") and name_node.contents is not None:
length = len(name_node.contents)
if length:
start_c = start_col - 1
end_c = start_c + length
if start_c <= char_zero_based <= end_c:
self.is_decl = True
finder = _DeclFinder()
tree.accept(finder, recurse=True)
return finder.is_decl
+17 -9
View File
@@ -1,17 +1,17 @@
import enum import enum
from typing import List from typing import List
from tclint.syntax_tree import Visitor, QuotedWord, Command, BareWord
from tclint.commands.plugins import PluginManager
import attrs import attrs
from common.load_data import standard_items from common.load_data import standard_items
import lsprotocol.types as lsp from tclint.commands.plugins import PluginManager
from tclint.syntax_tree import BareWord, Command, QuotedWord, Visitor
# Constructing a PluginManager scans entry points, and get_commands() rebuilds # Constructing a PluginManager scans entry points, and get_commands() rebuilds
# the builtin command set on every call. Semantic tokens are requested often, so # the builtin command set on every call. Semantic tokens are requested often, so
# cache the manager and the resolved commands per plugin set. # cache the manager and the resolved commands per plugin set.
_PLUGIN_MANAGER = None _PLUGIN_MANAGER = None
_COMMANDS_CACHE = {} _COMMANDS_CACHE = {}
_STANDARD_PROC_NAMES = frozenset(item.label for item in standard_items.nx_procs)
def _load_commands(plugins): def _load_commands(plugins):
@@ -60,13 +60,23 @@ TOKEN_TYPES = [
"string", "string",
"parameter", "parameter",
] ]
TOKEN_TYPE_INDEX = {}
for _token_index, _token_name in enumerate(TOKEN_TYPES):
TOKEN_TYPE_INDEX.setdefault(_token_name, _token_index)
class _Highlighter(Visitor): class _Highlighter(Visitor):
def __init__(self, plugins, custom_functions: dict[str : list[lsp.CompletionItem]]): def __init__(self, plugins, custom_functions):
self._commands = _load_commands(plugins) self._commands = _load_commands(plugins)
self._tokens = [] self._tokens = []
self.custom_functions = custom_functions if isinstance(custom_functions, dict):
self._custom_function_names = frozenset(
item.label
for items in custom_functions.values()
for item in items
)
else:
self._custom_function_names = frozenset(custom_functions)
def _append_token(self, position, length: int, tok_type: str, modifiers: List[TokenModifier] | None = None): def _append_token(self, position, length: int, tok_type: str, modifiers: List[TokenModifier] | None = None):
if position is None or length <= 0: if position is None or length <= 0:
@@ -129,9 +139,7 @@ class _Highlighter(Visitor):
# Highlight functions (custom or standard) when used as the routine # Highlight functions (custom or standard) when used as the routine
name = getattr(routine, "contents", None) name = getattr(routine, "contents", None)
if name: if name:
in_custom = any(item.label == name for items in self.custom_functions.values() for item in items) if name in self._custom_function_names or name in _STANDARD_PROC_NAMES:
in_standard = any(item.label == name for item in standard_items.nx_procs)
if in_custom or in_standard:
line, col = routine.contents_pos line, col = routine.contents_pos
self._append_token((line - 1, col - 1), len(name), "function", []) self._append_token((line - 1, col - 1), len(name), "function", [])
@@ -3,17 +3,15 @@ from concurrent.futures import ThreadPoolExecutor
from pathlib import Path from pathlib import Path
from threading import Event from threading import Event
THIS_DIR = Path(__file__).parent THIS_DIR = Path(__file__).parent
SRC_DIR = THIS_DIR.parent.parent / "src" SRC_DIR = THIS_DIR.parent.parent / "src"
if str(SRC_DIR) not in sys.path: if str(SRC_DIR) not in sys.path:
sys.path.insert(0, str(SRC_DIR)) sys.path.insert(0, str(SRC_DIR))
import lsprotocol.types as lsp # type: ignore
from pygls.workspace.text_document import TextDocument
import lsp_server import lsp_server
import lsprotocol.types as lsp # type: ignore
from lsp_tclserver import TclLanguageServer from lsp_tclserver import TclLanguageServer
from pygls.workspace.text_document import TextDocument
def _server() -> TclLanguageServer: def _server() -> TclLanguageServer:
@@ -63,6 +61,7 @@ def test_close_replaces_unsaved_index_with_saved_file(tmp_path: Path, monkeypatc
server = _server() server = _server()
server.update_poco_completion_for_file(document) server.update_poco_completion_for_file(document)
server.get_tree(document) server.get_tree(document)
server.get_lines(document)
monkeypatch.setattr(lsp_server, "LSP_SERVER", server) monkeypatch.setattr(lsp_server, "LSP_SERVER", server)
lsp_server.did_close( lsp_server.did_close(
@@ -75,6 +74,7 @@ def test_close_replaces_unsaved_index_with_saved_file(tmp_path: Path, monkeypatc
assert "unsaved_proc" not in signatures[document.path] assert "unsaved_proc" not in signatures[document.path]
assert "saved_proc" in signatures[document.path] assert "saved_proc" in signatures[document.path]
assert all(key[0] != document.uri for key in server._ast_cache) assert all(key[0] != document.uri for key in server._ast_cache)
assert all(key[0] != document.uri for key in server._line_cache)
def test_delete_and_rename_notifications_update_index(tmp_path: Path, monkeypatch): def test_delete_and_rename_notifications_update_index(tmp_path: Path, monkeypatch):
@@ -218,3 +218,52 @@ def test_delete_invalidates_in_flight_diagnostics(tmp_path: Path, monkeypatch):
future.result(timeout=5) future.result(timeout=5)
assert server.diagnostic_snapshot(document.uri) is None assert server.diagnostic_snapshot(document.uri) is None
def test_rapid_changes_analyze_only_latest_snapshot(tmp_path: Path, monkeypatch):
server = _server()
path = tmp_path / "debounced.tcl"
first = _document(path, "set value 1", version=1)
latest = _document(path, "set value 2", version=2)
calls = []
completed = Event()
monkeypatch.setattr(
server,
"compute_diagnostics",
lambda document: calls.append(("diagnostics", document.version)),
)
def record_index(document):
calls.append(("index", document.version))
completed.set()
return True
monkeypatch.setattr(server, "update_poco_completion_for_file", record_index)
server.schedule_document_analysis(first, delay_seconds=0.05)
server.schedule_document_analysis(latest, delay_seconds=0.05)
assert completed.wait(timeout=2)
assert calls == [("diagnostics", 2), ("index", 2)]
def test_variable_and_workspace_request_caches_are_reused(tmp_path: Path):
server = _server()
document = _document(
tmp_path / "cached.tcl",
"proc cached_proc {argument} { set local_value $argument }",
)
assert server.update_poco_completion_for_file(document)
first_variables = server.variable_index_for_document(document)
second_variables = server.variable_index_for_document(document)
first_completions = server.completion_items_snapshot()
second_completions = server.completion_items_snapshot()
first_names = server.custom_function_names_snapshot()
second_names = server.custom_function_names_snapshot()
assert first_variables is second_variables
assert first_completions is second_completions
assert first_names is second_names
assert "cached_proc" in first_names
@@ -0,0 +1,209 @@
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 tools.inlay_hint import (
InlayHintGenerator,
InlayHintParameter,
InlayHintSignature,
build_builtin_inlay_signatures,
build_custom_inlay_signatures,
)
from tools.navigation import build_file_symbol_index
from tools.parser import CustomParser
def _signature(*names: str, variadic: bool = False) -> InlayHintSignature:
parameters = tuple(
InlayHintParameter(
name=name,
variadic=variadic and index == len(names) - 1,
)
for index, name in enumerate(names)
)
return InlayHintSignature(
parameters=parameters,
display_label=" ".join(["test_proc", *names]),
)
def _generate(
source: str,
signatures: dict[str, InlayHintSignature],
**options,
) -> list[lsp.InlayHint]:
tree = CustomParser().parse(source)
generator = InlayHintGenerator(source, signatures, **options)
return generator.generate(tree)
def _labels(hints: list[lsp.InlayHint]) -> list[str]:
labels = []
for hint in hints:
if isinstance(hint.label, str):
labels.append(hint.label)
else:
labels.append("".join(part.value for part in hint.label))
return labels
def test_many_parameters_are_returned_without_server_side_truncation():
names = tuple(f"parameter_{index}" for index in range(10))
source = "test_proc " + " ".join(str(index) for index in range(10))
hints = _generate(source, {"test_proc": _signature(*names)})
assert _labels(hints) == [f"{name}:" for name in names]
assert all("" not in label and "..." not in label for label in _labels(hints))
def test_only_hints_inside_requested_range_are_returned():
source = "test_proc first\nset spacer 1\ntest_proc second"
requested_range = lsp.Range(
start=lsp.Position(line=2, character=0),
end=lsp.Position(line=3, character=0),
)
hints = _generate(
source,
{"test_proc": _signature("value")},
requested_range=requested_range,
)
assert len(hints) == 1
assert hints[0].position.line == 2
def test_range_walk_keeps_nested_commands_inside_proc_body():
source = "proc wrapper {} {\n test_proc nested\n}"
requested_range = lsp.Range(
start=lsp.Position(line=1, character=0),
end=lsp.Position(line=2, character=0),
)
hints = _generate(
source,
{"test_proc": _signature("value")},
requested_range=requested_range,
)
assert _labels(hints) == ["value:"]
assert hints[0].position.line == 1
def test_matching_variable_name_can_be_suppressed():
source = "test_proc $value $other"
signature = _signature("value", "result")
suppressed = _generate(source, {"test_proc": signature})
visible = _generate(
source,
{"test_proc": signature},
suppress_when_argument_matches_name=False,
)
assert _labels(suppressed) == ["result:"]
assert _labels(visible) == ["value:", "result:"]
def test_literal_mode_hides_variable_argument_hints():
source = 'test_proc $value "literal"'
hints = _generate(
source,
{"test_proc": _signature("first", "second")},
parameter_names="literals",
)
assert _labels(hints) == ["second:"]
def test_variadic_parameter_labels_every_remaining_argument():
hints = _generate(
"test_proc first second third fourth",
{"test_proc": _signature("required", "args", variadic=True)},
)
assert _labels(hints) == ["required:", "args:", "args:", "args:"]
def test_builtin_signature_has_variadic_hints_and_parameter_documentation():
signatures = build_builtin_inlay_signatures(
[
{
"label": "MOM_force",
"description": "Controls address output.",
"format": "MOM_force <mode> <address_1 ... address_n>",
"parameters": [
{"name": "mode", "desc": "Output mode."},
{
"name": "address_1 ... address_n",
"desc": "Output addresses.",
},
],
}
]
)
hints = _generate("MOM_force Always X Y", signatures)
assert _labels(hints) == ["mode:", "address:", "address:"]
assert hints[1].label[0].tooltip == "Output addresses." # type: ignore[index]
assert hints[0].tooltip.value.endswith("Controls address output.") # type: ignore[union-attr]
def test_current_file_signature_and_definition_location_take_priority(tmp_path: Path):
other_path = tmp_path / "other.tcl"
current_path = tmp_path / "current.tcl"
other_source = "proc shared {from_other} { return $from_other }"
current_source = "proc shared {from_current args} { return $from_current }"
parser = CustomParser()
other_tree = parser.parse(other_source)
current_tree = parser.parse(current_source)
indexes = {
str(other_path): build_file_symbol_index(
str(other_path), other_path.as_uri(), other_tree
),
str(current_path): build_file_symbol_index(
str(current_path), current_path.as_uri(), current_tree
),
}
signatures = build_custom_inlay_signatures(
{
str(current_path): {"shared": ["from_current", "args"]},
str(other_path): {"shared": ["from_other"]},
},
{
str(current_path): {"shared": "Current documentation."},
str(other_path): {"shared": "Other documentation."},
},
indexes,
str(current_path),
)
signature = signatures["shared"]
assert [parameter.name for parameter in signature.parameters] == [
"from_current",
"args",
]
assert signature.parameters[-1].variadic
assert signature.documentation == "Current documentation."
assert signature.location is not None
assert signature.location.uri == current_path.as_uri()
def test_positions_use_lsp_utf16_offsets():
source = 'test_proc "😀" second'
hints = _generate(
source,
{"test_proc": _signature("first", "second")},
)
assert hints[1].position.character == source.index("second") + 1