feat: use Protocol for config change listener and support passing parameters

Author: observerw

src/lsp_client/capability/server_request/configuration.py

@@ -80,6 +80,26 @@ def register_server_request_hooks(
         self, registry: ServerRequestHookRegistry
     ) -> None:
         super().register_server_request_hooks(registry)
+
+        # Automatically bind change notification if both capabilities are present
+        from lsp_client.capability.notification.did_change_configuration import (
+            WithNotifyDidChangeConfiguration,
+        )
+
+        if self.configuration_map and isinstance(
+            self, WithNotifyDidChangeConfiguration
+        ):
+            # We use a lambda to avoid sync/async issues if the notification
+            # needs to be scheduled on an event loop
+            import asyncer
+
+            def on_config_change(config_map: ConfigurationMap, **kwargs: Any):
+                with logger.contextualize(method="didChangeConfiguration"):
+                    logger.debug("Configuration changed, notifying server")
+                    asyncer.runnify(self.notify_change_configuration)()
+
+            self.configuration_map.on_change(on_config_change)
+
         registry.register(
             lsp_type.WORKSPACE_CONFIGURATION,
             ServerRequestHook(

src/lsp_client/utils/config.py

@@ -2,7 +2,7 @@
 
 import fnmatch
 from copy import deepcopy
-from typing import Any
+from typing import Any, Protocol, runtime_checkable
 
 from attrs import define, field
 from loguru import logger
@@ -23,6 +23,15 @@ def deep_merge(base: dict[str, Any], update: dict[str, Any]) -> dict[str, Any]:
     return result
 
 
+@runtime_checkable
+class ConfigurationChangeListener(Protocol):
+    """
+    Protocol for configuration change listeners.
+    """
+
+    def __call__(self, config_map: ConfigurationMap, **kwargs: Any) -> Any: ...
+
+
 @define
 class ConfigurationMap:
     """
@@ -32,21 +41,49 @@ class ConfigurationMap:
 
     _global_config: dict[str, Any] = field(factory=dict)
     _scoped_configs: list[tuple[str, dict[str, Any]]] = field(factory=list)
+    _on_change_callbacks: list[ConfigurationChangeListener] = field(
+        factory=list, init=False
+    )
+
+    def on_change(self, callback: ConfigurationChangeListener) -> None:
+        """
+        Register a callback to be called when the configuration changes.
+        """
+        self._on_change_callbacks.append(callback)
+
+    def _notify_change(self, **kwargs: Any) -> None:
+        for callback in self._on_change_callbacks:
+            try:
+                callback(self, **kwargs)
+            except Exception as e:
+                logger.error(f"Error in configuration change callback: {e}")
+
+    def update_global(
+        self, config: dict[str, Any], merge: bool = True, **kwargs: Any
+    ) -> None:
+        """
+        Update global configuration.
+        """
+        if merge:
+            self._global_config = deep_merge(self._global_config, config)
+        else:
+            self._global_config = deepcopy(config)
+        self._notify_change(**kwargs)
 
-    def add_scope(self, pattern: str, config: dict[str, Any]) -> None:
+    def add_scope(self, pattern: str, config: dict[str, Any], **kwargs: Any) -> None:
         """
         Add a configuration override for a specific file pattern.
 
         :param pattern: Glob pattern (e.g. "**/tests/**", "*.py")
         :param config: The configuration dict to merge for this scope
         """
         self._scoped_configs.append((pattern, config))
+        self._notify_change(**kwargs)
 
     def _get_section(self, config: Any, section: str | None) -> Any:
         if not section:
             return config
 
-        # Traverse the config dictionary using the section path (e.g. "python.analysis")
         current = config
         for part in section.split("."):
             if isinstance(current, dict) and part in current:
@@ -56,10 +93,8 @@ def _get_section(self, config: Any, section: str | None) -> Any:
         return current
 
     def get(self, scope_uri: str | None, section: str | None) -> Any:
-        # Start with global config
         final_config = self._global_config
 
-        # If we have a scope, merge matching scoped configs
         if scope_uri:
             try:
                 path_str = str(from_local_uri(scope_uri))