Refactor LSP manager architecture and event + completion pipeline

- Replace legacy LSPManagerClient + LSPManagerUI with ClientManager and UIManager
- Remove WebsocketClient in favor of unified Websocket implementation
- Move LSP initialization config loading into centralized config module
- Update LSPClient to support dynamic socket, improved init params, and doc version tracking
- Introduce range-based didChange notifications and add implementation + references requests
- Expand LSPClientEvents to support implementation/references and structured range edits
- Simplify websocket response handling and normalize LSP response parsing flow
- Decouple UI from LSP manager core; UI now emits address/port for client creation
- Refactor completion provider pipeline:
  - Split TextChangedEvent into TextInsertedEvent and DeleteRangeEvent
  - Update ProviderResponseCacheBase and controller dispatch paths accordingly
- Improve SourceBuffer and SourceFile event tracking with delete-range support
- Update plugin system:
  - Centralize command/provider registration via helper methods
  - Add LSP commands: definition, references, implementation, toggle UI
- Enhance response handlers to support references and implementation hooks
- Improve Python LSP config (jedi completion, signatures, references enabled)
- Fix minor GTK lifecycle and buffer signal handling issues
- Clean up unused imports, dead code, and outdated JSON server configs
This commit is contained in:
2026-04-11 15:36:59 -05:00
parent 0dc21cbb82
commit a8ad015e05
34 changed files with 896 additions and 575 deletions

View File

@@ -0,0 +1,191 @@
# Python imports
# Lib imports
# Application imports
from libs.event_factory import Code_Event_Types
class ClientManagerEventsMixin:
def _get_controller(self, lang_id, action: str):
controller = self.clients.get(lang_id)
if not controller:
logger.debug(f"No LSP client for '{lang_id}', skipping {action}...")
return controller
def _uri(self, fpath: str) -> str:
return fpath if fpath.startswith("file://") else f"file://{fpath}"
def _text(self, buffer, *, hidden = False):
return buffer.get_text(
*buffer.get_bounds(),
include_hidden_chars=hidden
)
def _version(self, controller, uri, bump = False):
if bump:
controller.doc_vers[uri] = controller.doc_vers.get(uri, -1) + 1
return controller.doc_vers.get(uri, 0)
def _activate(self, lang_id):
self.active_language_id = lang_id
def process_file_load(self, event: Code_Event_Types.AddedNewFileEvent):
f = event.file
if not (c := self._get_controller(f.ftype, "didOpen")): return
uri = self._uri(f.fpath)
self._activate(f.ftype)
c._lsp_did_open({
"uri": uri,
"language_id": f.ftype,
"text": self._text(f.buffer),
})
def process_file_close(self, event: Code_Event_Types.RemovedFileEvent):
f = event.file
if not (c := self._get_controller(f.ftype, "didClose")): return
uri = self._uri(f.fpath)
c.doc_vers.pop(uri, None)
c._lsp_did_close({"uri": uri})
def process_file_save(self, event: Code_Event_Types.SavedFileEvent):
f = event.file
if not (c := self._get_controller(f.ftype, "didSave")): return
uri = self._uri(f.fpath)
self._activate(f.ftype)
c._lsp_did_save({
"uri": uri,
"text": self._text(f.buffer),
})
def process_file_change(self, event: Code_Event_Types.TextChangedEvent):
f = event.file
if not (c := self._get_controller(f.ftype, "didChange")): return
uri = self._uri(f.fpath)
self._activate(f.ftype)
version = self._version(c, uri, bump = True)
c._lsp_did_change({
"uri": uri,
"language_id": f.ftype,
"version": version,
"text": self._text(f.buffer, hidden = True),
})
it = f.buffer.get_iter_at_mark(f.buffer.get_insert())
self._set_cache_refresh_trigger(
f.ftype, f.fpath,
it.get_line(),
it.get_line_offset() + 1
)
def _iter_pos(self, it):
return it.get_line(), it.get_line_offset()
def process_file_text_inserted(self, event: Code_Event_Types.TextInsertedEvent):
f = event.file
if not (c := self._get_controller(f.ftype, "didChange")): return
uri = self._uri(f.fpath)
self._activate(f.ftype)
start_it = event.location.copy()
end_it = event.location.copy()
if event.length > 1:
start_it.backward_chars(event.length)
sl, sc = self._iter_pos(start_it)
el, ec = self._iter_pos(end_it)
sc -= 0 if event.length > 1 else 1
ec -= 1
version = self._version(c, uri, bump = True)
c._lsp_did_change_range({
"uri": uri,
"language_id": f.ftype,
"version": version,
"text": event.text,
"line": sl,
"column": sc,
"end_line": el,
"end_column": ec,
})
it = event.buffer.get_iter_at_mark(event.buffer.get_insert())
self._set_cache_refresh_trigger(
f.ftype, f.fpath, *self._iter_pos(it)
)
def process_file_delete_range(self, event: Code_Event_Types.DeleteRangeEvent):
f = event.file
if not (c := self._get_controller(f.ftype, "didChange")): return
uri = self._uri(f.fpath)
self._activate(f.ftype)
start_it, end_it = event.start.copy(), event.end.copy()
if start_it.compare(end_it) > 0:
start_it, end_it = end_it, start_it
sl, sc = self._iter_pos(start_it)
el, ec = self._iter_pos(end_it)
version = self._version(c, uri, bump = True)
c._lsp_did_change_range({
"uri": uri,
"language_id": f.ftype,
"version": version,
"text": "",
"line": sl,
"column": sc,
"end_line": el,
"end_column": ec,
})
it = event.buffer.get_iter_at_mark(event.buffer.get_insert())
self._set_cache_refresh_trigger(
f.ftype, f.fpath, *self._iter_pos(it)
)
def _request(self, method, lang_id, fpath, **extra):
if not (c := self._get_controller(lang_id, method)): return
uri = self._uri(fpath)
self._activate(lang_id)
payload = {
"uri": uri,
"language_id": lang_id,
"version": self._version(c, uri),
**extra
}
getattr(c, method)(payload)
def process_definition(self, lang_id, fpath, line, column):
self._request("_lsp_definition", lang_id, fpath, line = line, column = column)
def process_implementation_definition(self, lang_id, fpath, line, column):
self._request("_lsp_implementation", lang_id, fpath, line = line, column = column)
def process_references_definition(self, lang_id, fpath, line, column):
self._request("_lsp_references", lang_id, fpath, line = line, column = column)
def process_completion_request(self, lang_id, fpath, line, column):
self._request("_lsp_completion", lang_id, fpath, line = line, column = column)
def _set_cache_refresh_trigger(self, lang_id, fpath, line, column):
self.process_completion_request(lang_id, fpath, line, column)

View File

@@ -1,144 +0,0 @@
# Python imports
# Lib imports
import gi
from gi.repository import GLib
# Application imports
from libs.event_factory import Code_Event_Types
class LSPClientEventsMixin:
def process_file_load(self, event: Code_Event_Types.AddedNewFileEvent):
lang_id = event.file.ftype
if lang_id not in self.clients:
logger.debug(f"No LSP client for '{lang_id}', skipping didOpen")
return
controller = self.clients[lang_id]
fpath = event.file.fpath
uri = f"file://{fpath}" if not fpath.startswith("file://") else fpath
buffer = event.file.buffer
text = buffer.get_text(*buffer.get_bounds())
self.active_language_id = lang_id
controller._lsp_did_open({
"uri": uri,
"language_id": lang_id,
"text": text
})
def process_file_close(self, event: Code_Event_Types.RemovedFileEvent):
lang_id = event.file.ftype
if lang_id not in self.clients:
logger.debug(f"No LSP client for '{lang_id}', skipping didClose")
return
controller = self.clients[lang_id]
fpath = event.file.fpath
uri = f"file://{fpath}" if not fpath.startswith("file://") else fpath
controller._lsp_did_close({"uri": uri})
def process_file_save(self, event: Code_Event_Types.SavedFileEvent):
lang_id = event.file.ftype
if lang_id not in self.clients:
logger.debug(f"No LSP client for '{lang_id}', skipping didSave")
return
controller = self.clients[lang_id]
fpath = event.file.fpath
uri = f"file://{fpath}" if not fpath.startswith("file://") else fpath
buffer = event.file.buffer
text = buffer.get_text(*buffer.get_bounds())
self.active_language_id = lang_id
controller._lsp_did_save({"uri": uri, "text": text})
def process_file_change(self, event: Code_Event_Types.TextChangedEvent):
self._clear_delayed_cache_refresh_trigger()
lang_id = event.file.ftype
if lang_id not in self.clients:
logger.debug(f"No LSP client for '{lang_id}', skipping didChange")
return
controller = self.clients[lang_id]
fpath = event.file.fpath
uri = f"file://{fpath}" if not fpath.startswith("file://") else fpath
buffer = event.file.buffer
text = buffer.get_text(*buffer.get_bounds())
self.active_language_id = lang_id
controller._lsp_did_change({
"uri": uri,
"language_id": lang_id,
"version": 1,
"text": text
})
iter = buffer.get_iter_at_mark( buffer.get_insert() )
line = iter.get_line()
column = iter.get_line_offset()
self._set_cache_refresh_trigger(
lang_id, fpath, line, column
)
def process_goto_definition(
self, lang_id: str, fpath: str, line: int, column: int
):
if lang_id not in self.clients:
logger.debug(f"No LSP client for '{lang_id}', skipping goto definition")
return
controller = self.clients[lang_id]
uri = f"file://{fpath}" if not fpath.startswith("file://") else fpath
self.active_language_id = lang_id
controller._lsp_definition({
"uri": uri,
"language_id": lang_id,
"version": 1,
"line": line,
"column": column
})
def process_completion_request(
self, lang_id: str, fpath: str, line: int, column: int
):
if lang_id not in self.clients:
logger.debug(f"No LSP client for '{lang_id}', skipping completion")
return
controller = self.clients[lang_id]
uri = f"file://{fpath}" if not fpath.startswith("file://") else fpath
self.active_language_id = lang_id
controller._lsp_completion({
"uri": uri,
"language_id": lang_id,
"version": 1,
"line": line,
"column": column
})
def _clear_delayed_cache_refresh_trigger(self):
if self._cache_refresh_timeout_id:
GLib.source_remove(self._cache_refresh_timeout_id)
def _set_cache_refresh_trigger(
self, lang_id: str, fpath: str, line: int, column: int
):
def trigger_cache_refresh(lang_id, fpath, line, column):
self._cache_refresh_timeout_id = None
self.process_completion_request(
lang_id, fpath, line, column
)
return False
self._cache_refresh_timeout_id = GLib.timeout_add(1500, trigger_cache_refresh, lang_id, fpath, line, column)

View File

@@ -0,0 +1,72 @@
# Python imports
import json
# Lib imports
import gi
from gi.repository import GObject
# Application imports
class UIManagerClientsMixin:
def _create_client(self, widget, sibling):
if not self.source_view: return
buffer = self.source_view.get_buffer()
lang_id = self.combo_box.get_active_text()
if not lang_id: return
workspace_dir = self.path_entry.get_text()
self.emit('create-client', lang_id, workspace_dir)
def _close_client(self, widget, sibling):
lang_id = self.combo_box.get_active_text()
if not lang_id: return
self.emit('close-client', lang_id)
def set_source_view_text(self, workspace_dir: str):
lang_id = self.combo_box.get_active_text()
if not lang_id: return
json_str = self.client_configs[lang_id]\
.replace("{workspace.folder}", workspace_dir)\
.replace("{user.home}", self._USER_HOME)
self.source_view.get_buffer().set_text(json_str, -1)
def add_client_listing(self, lang_id: str, lang_config: str):
self.combo_box.append_text(lang_id)
self.client_configs[lang_id] = lang_config
def remove_client_listing(self, lang_id: str):
model = self.combo_box.get_model()
for i, row in enumerate(model):
if row[0] == lang_id:
self.combo_box.remove(i)
break
self.client_configs.pop(lang_id, None)
def toggle_client_buttons(self, show_close: bool):
self.create_client_bttn.set_visible(not show_close)
self.close_client_bttn.set_visible(show_close)
def get_init_opts(self, lang_id: str) -> dict:
if not lang_id or lang_id not in self.client_configs: return {}
try:
buffer = self.source_view.get_buffer()
json_str = buffer.get_text(*buffer.get_bounds(), -1)
lang_config = json.loads(json_str)
except json.JSONDecodeError as e:
logger.error(f"Invalid JSON for {lang_id}: {e}")
return {}
return lang_config.get("initialization-options", {})

View File

@@ -0,0 +1,77 @@
# Python imports
# Lib imports
import gi
gi.require_version("Gtk", "3.0")
from gi.repository import GLib
from gi.repository import Gtk
# Application imports
class UIManagerEventsMixin:
def _setup_signals(self):
self.connect("show", self._handle_show)
self.connect("destroy", self._handle_destroy)
def _subscribe_to_events(self):
...
def _handle_show(self, widget):
GLib.idle_add(self.path_entry.grab_focus)
def _handle_destroy(self, widget):
self.disconnect_by_func(self._handle_show)
self.disconnect_by_func(self._handle_destroy)
self.path_bttn.disconnect_by_func(self._file_set)
self.combo_box.disconnect_by_func(self._on_combo_changed)
self.hide_bttn.disconnect(self.hide_bttn_id)
self.create_client_bttn.disconnect_by_func(self._create_client)
self.close_client_bttn.disconnect_by_func(self._close_client)
def _map_resize(self, widget, parent):
parent_x, \
parent_y = parent.get_position()
parent_width, \
parent_height = parent.get_size()
if parent_width == 0 or parent_height == 0: return
width = int(parent_width * 0.75)
height = int(parent_height * 0.75)
widget.resize(width, height)
x = parent_x + (parent_width - width) // 2
y = parent_y + (parent_height - height) // 2
widget.move(x, y)
def _path_changed(self, widget, buttons_widget):
if not widget.get_text():
self.path_bttn.unselect_all()
self.path_bttn.emit("file-set")
buttons_widget.hide()
return
self.set_source_view_text( self.path_entry.get_text() )
buttons_widget.show()
def _file_set(self, widget):
fname = widget.get_filename()
fname = "" if not fname else fname
self.path_entry.set_text(fname)
lang_id = self.combo_box.get_active_text()
if not lang_id or lang_id not in self.client_configs: return
self.set_source_view_text(
"{workspace.folder}" if not fname else fname
)
def _on_combo_changed(self, combo: Gtk.ComboBoxText):
lang_id = combo.get_active_text()
self.set_source_view_text( self.path_entry.get_text() )

View File

@@ -0,0 +1,98 @@
# Python imports
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
gi.require_version('GtkSource', '4')
from gi.repository import Gtk
from gi.repository import GtkSource
# Application imports
from ..config import get_lsp_host_addr, get_lsp_host_port
class UIManagerSetupMixin:
def _setup_styling(self):
self.set_modal(True)
self.set_decorated(False)
self.set_vexpand(True)
self.set_hexpand(True)
def _load_widgets(self):
content_area = self.get_content_area()
self.main_box = Gtk.Grid()
self.path_entry = Gtk.SearchEntry()
self.path_bttn = Gtk.FileChooserButton.new(
title = "Workspace Folder",
action = Gtk.FileChooserAction.SELECT_FOLDER
)
self.combo_box = Gtk.ComboBoxText()
self.hide_bttn = Gtk.Button(label = "X")
self.adddress_entry = Gtk.Entry()
adjustment = Gtk.Adjustment(
value = get_lsp_host_port(),
lower = 1,
upper = 65535,
step_increment = 1,
page_increment = 10,
page_size = 0
)
self.adddress_port = Gtk.SpinButton()
self.adddress_port.set_adjustment(adjustment)
self.adddress_port.set_digits(0) # integers only
bttn_box = Gtk.Box()
self.create_client_bttn = Gtk.Button(label = "Create Language Client")
self.close_client_bttn = Gtk.Button(label = "Close Language Client")
self.path_entry.set_can_focus(False)
self.path_entry.set_placeholder_text("Workspace Folder...")
self.path_entry.connect("changed", self._path_changed, bttn_box)
self.path_bttn.set_halign(Gtk.Align.FILL)
self.adddress_entry.set_placeholder_text("Address...")
self.adddress_entry.set_text( get_lsp_host_addr() )
self.path_bttn.connect("file-set", self._file_set)
self.combo_box.connect("changed", self._on_combo_changed)
self.hide_bttn_id = self.hide_bttn.connect("clicked", lambda widget: self.hide())
self.create_client_bttn.connect("clicked", self._create_client, self.close_client_bttn)
self.close_client_bttn.connect("clicked", self._close_client, self.create_client_bttn)
self.main_box.set_column_spacing(15)
self.main_box.set_row_spacing(15)
bttn_box.pack_start(self.create_client_bttn, False, False, 0)
bttn_box.pack_start(self.close_client_bttn, False, False, 0)
self.main_box.attach(child = self.path_entry, left = 0, top = 0, width = 4, height = 1)
self.main_box.attach(child = self.path_bttn, left = 4, top = 0, width = 1, height = 1)
self.main_box.attach(child = self.combo_box, left = 5, top = 0, width = 1, height = 1)
self.main_box.attach(child = self.hide_bttn, left = 6, top = 0, width = 1, height = 1)
self.main_box.attach(child = self.adddress_entry, left = 0, top = 1, width = 2, height = 1)
self.main_box.attach(child = self.adddress_port, left = 2, top = 1, width = 2, height = 1)
self.main_box.attach(child = bttn_box, left = 4, top = 1, width = 3, height = 1)
content_area.set_vexpand(True)
content_area.set_hexpand(True)
content_area.add(self.main_box)
content_area.show_all()
self.close_client_bttn.hide()
bttn_box.hide()
def set_source_view(self, scrolled_win, source_view):
lang_manager = GtkSource.LanguageManager()
buffer = source_view.get_buffer()
language = lang_manager.get_language("json")
self.source_view = source_view
buffer.set_language(language)
buffer.set_style_scheme(self.source_view.syntax_theme)
self.main_box.attach(scrolled_win, 0, 2, 7, 1)