temp/src/cleveragents/lsp/runtime.py

"""Functional LSP Runtime for managing language server processes.

The ``LspRuntime`` manages LSP server processes and provides code
intelligence (diagnostics, completions) to actors.  It delegates
process lifecycle to :class:`LspLifecycleManager` and protocol
communication to :class:`LspClient`.

When a server is started, the runtime:

1. Looks up the :class:`LspServerConfig` from the :class:`LspRegistry`.
2. Delegates to the lifecycle manager to spawn (or reuse) the process.
3. Tracks running servers for query operations.

When a query (diagnostics, completions) is issued, the runtime:

1. Verifies the server is running and healthy.
2. If crashed, attempts automatic restart.
3. Delegates to the :class:`LspClient` for the protocol call.

Based on ``docs/specification.md`` LSP Server Lifecycle (lines 20744-20758)
and issue #826.
"""

from __future__ import annotations

import contextlib
import os
from typing import Any

import structlog

from cleveragents.lsp.errors import LspError, LspServerNotFoundError
from cleveragents.lsp.lifecycle import LspLifecycleManager
from cleveragents.lsp.registry import LspRegistry

logger = structlog.get_logger(__name__)


class LspRuntime:
    """Functional runtime for LSP servers.

    Manages the full lifecycle of language servers: start, query,
    health-check, restart, and stop.

    Args:
        registry: The LSP server registry to look up configurations.
        lifecycle_manager: Optional lifecycle manager (created if not
            provided).
    """

    def __init__(
        self,
        registry: LspRegistry | None = None,
        lifecycle_manager: LspLifecycleManager | None = None,
    ) -> None:
        self._registry = registry or LspRegistry()
        self._lifecycle = lifecycle_manager or LspLifecycleManager()

    @property
    def registry(self) -> LspRegistry:
        """The underlying LSP server registry."""
        return self._registry

    @property
    def lifecycle(self) -> LspLifecycleManager:
        """The underlying lifecycle manager."""
        return self._lifecycle

    def start_server(self, name: str, workspace_path: str) -> None:
        """Start an LSP server for *name* in *workspace_path*.

        Looks up the server configuration from the registry and
        delegates to the lifecycle manager to spawn (or reuse) the
        process.  The LSP ``initialize`` handshake is performed
        automatically.

        Args:
            name: Namespaced server name (must be registered).
            workspace_path: Root directory for the language server.

        Raises:
            ValueError: If *name* or *workspace_path* is empty.
            LspServerNotFoundError: If *name* is not in the registry.
            LspError: If the server process cannot be started.
        """
        if not name:
            raise ValueError("name must be a non-empty string")
        if not workspace_path:
            raise ValueError("workspace_path must be a non-empty string")

        config = self._registry.get_or_raise(name)

        logger.info(
            "lsp.runtime.starting_server",
            server=name,
            workspace=workspace_path,
            command=config.command,
        )

        self._lifecycle.start_server(config, workspace_path)

    def stop_server(self, name: str) -> None:
        """Stop the LSP server identified by *name*.

        Releases one reference to the server.  The process is only
        terminated when all references are released.

        Args:
            name: Namespaced server name.

        Raises:
            ValueError: If *name* is empty.
            LspServerNotFoundError: If the server is not running.
        """
        if not name:
            raise ValueError("name must be a non-empty string")

        logger.info("lsp.runtime.stopping_server", server=name)
        self._lifecycle.stop_server(name)

    def get_diagnostics(self, name: str, file_path: str) -> list[Any]:
        """Retrieve diagnostics for *file_path* from the named server.

        Opens the file, sends ``textDocument/didOpen`` to the server,
        reads cached diagnostics (pushed asynchronously by the server),
        then closes the document.

        Args:
            name: Namespaced server name.
            file_path: Path to the file being diagnosed.

        Returns:
            List of LSP Diagnostic dicts.

        Raises:
            ValueError: If *name* or *file_path* is empty.
            LspServerNotFoundError: If the server is not running.
            LspError: If the server has crashed.
        """
        if not name:
            raise ValueError("name must be a non-empty string")
        if not file_path:
            raise ValueError("file_path must be a non-empty string")

        client = self._get_healthy_client(name)
        uri = self._path_to_uri(file_path)

        # Open the file so the server analyses it
        try:
            text = self._read_file(file_path)
        except OSError as exc:
            raise LspError(
                f"Cannot read file for diagnostics: {file_path}",
                details={"file": file_path, "error": str(exc)},
            ) from exc

        language_id = self._detect_language(file_path)
        client.did_open(uri, language_id, version=1, text=text)

        # Give the server time to process and push diagnostics
        diagnostics = client.get_diagnostics(uri)

        # Close the document
        client.did_close(uri)

        logger.info(
            "lsp.runtime.diagnostics",
            server=name,
            file=file_path,
            count=len(diagnostics),
        )
        return diagnostics

    def get_completions(
        self,
        name: str,
        file_path: str,
        line: int,
        column: int,
    ) -> list[Any]:
        """Retrieve completions for a position in *file_path*.

        Args:
            name: Namespaced server name.
            file_path: Path to the file.
            line: 1-based line number.
            column: 1-based column number.

        Returns:
            List of CompletionItem dicts.

        Raises:
            ValueError: If inputs are invalid.
            LspServerNotFoundError: If the server is not running.
            LspError: If the server has crashed.
        """
        if not name:
            raise ValueError("name must be a non-empty string")
        if not file_path:
            raise ValueError("file_path must be a non-empty string")
        if line < 1:
            raise ValueError("line must be >= 1")
        if column < 1:
            raise ValueError("column must be >= 1")

        client = self._get_healthy_client(name)
        uri = self._path_to_uri(file_path)

        # Open the file
        try:
            text = self._read_file(file_path)
        except OSError as exc:
            raise LspError(
                f"Cannot read file for completions: {file_path}",
                details={"file": file_path, "error": str(exc)},
            ) from exc

        language_id = self._detect_language(file_path)
        client.did_open(uri, language_id, version=1, text=text)

        # LSP uses 0-based line/character
        completions = client.get_completions(uri, line - 1, column - 1)

        client.did_close(uri)

        logger.info(
            "lsp.runtime.completions",
            server=name,
            file=file_path,
            position=f"{line}:{column}",
            count=len(completions),
        )
        return completions

    def get_hover(
        self,
        name: str,
        file_path: str,
        line: int,
        column: int,
    ) -> dict[str, Any] | None:
        """Retrieve hover information for a position in *file_path*.

        Args:
            name: Namespaced server name.
            file_path: Path to the file.
            line: 1-based line number.
            column: 1-based column number.

        Returns:
            Hover result dict or ``None`` if no info is available.
        """
        if not name:
            raise ValueError("name must be a non-empty string")
        if not file_path:
            raise ValueError("file_path must be a non-empty string")
        if line < 1:
            raise ValueError("line must be >= 1")
        if column < 1:
            raise ValueError("column must be >= 1")

        client = self._get_healthy_client(name)
        uri = self._path_to_uri(file_path)

        try:
            text = self._read_file(file_path)
        except OSError as exc:
            raise LspError(
                f"Cannot read file for hover: {file_path}",
                details={"file": file_path, "error": str(exc)},
            ) from exc

        language_id = self._detect_language(file_path)
        client.did_open(uri, language_id, version=1, text=text)
        try:
            hover = client.get_hover(uri, line - 1, column - 1)
        finally:
            client.did_close(uri)

        logger.info(
            "lsp.runtime.hover",
            server=name,
            file=file_path,
            position=f"{line}:{column}",
            has_result=hover is not None,
        )
        return hover

    def get_definitions(
        self,
        name: str,
        file_path: str,
        line: int,
        column: int,
    ) -> list[dict[str, Any]]:
        """Retrieve go-to-definition locations for a position.

        Args:
            name: Namespaced server name.
            file_path: Path to the file.
            line: 1-based line number.
            column: 1-based column number.

        Returns:
            List of Location dicts.
        """
        if not name:
            raise ValueError("name must be a non-empty string")
        if not file_path:
            raise ValueError("file_path must be a non-empty string")
        if line < 1:
            raise ValueError("line must be >= 1")
        if column < 1:
            raise ValueError("column must be >= 1")

        client = self._get_healthy_client(name)
        uri = self._path_to_uri(file_path)

        try:
            text = self._read_file(file_path)
        except OSError as exc:
            raise LspError(
                f"Cannot read file for definitions: {file_path}",
                details={"file": file_path, "error": str(exc)},
            ) from exc

        language_id = self._detect_language(file_path)
        client.did_open(uri, language_id, version=1, text=text)
        try:
            definitions = client.get_definitions(uri, line - 1, column - 1)
        finally:
            client.did_close(uri)

        logger.info(
            "lsp.runtime.definitions",
            server=name,
            file=file_path,
            position=f"{line}:{column}",
            count=len(definitions),
        )
        return definitions

    def _get_healthy_client(self, name: str) -> Any:
        """Get the client for a server, restarting if crashed.

        Args:
            name: Server name.

        Returns:
            An :class:`LspClient` connected to a healthy server.

        Raises:
            LspServerNotFoundError: If the server was never started.
            LspError: If restart fails.
        """
        if not self._lifecycle.health_check(name):
            logger.warning("lsp.runtime.server_crashed", server=name)
            try:
                return self._lifecycle.restart_server(name)
            except (LspError, LspServerNotFoundError):
                raise

        return self._lifecycle.get_client(name)

    @staticmethod
    def _path_to_uri(file_path: str) -> str:
        """Convert a filesystem path to a ``file://`` URI."""
        if file_path.startswith("file://"):
            return file_path
        return f"file://{file_path}"

    @staticmethod
    def _read_file(file_path: str) -> str:
        """Read file contents as UTF-8 text.

        Raises:
            LspError: If the resolved path is a directory or device.
        """
        resolved = os.path.realpath(file_path)
        if not os.path.isfile(resolved):
            raise LspError(
                f"Not a regular file: {file_path}",
                details={"file": file_path, "resolved": resolved},
            )
        with open(resolved, encoding="utf-8") as f:
            return f.read()

    @staticmethod
    def _detect_language(file_path: str) -> str:
        """Detect the language ID from a file extension.

        Delegates to :class:`LanguageDiscovery` for the full 4-layer
        detection process.
        """
        from cleveragents.lsp.discovery import LanguageDiscovery

        return LanguageDiscovery().detect_file_language(file_path)

    def activate_bindings(
        self,
        bindings: list[Any],
        workspace_path: str,
    ) -> list[str]:
        """Start LSP servers required by actor compilation bindings.

        For each binding, looks up the server config in the registry
        and starts it.  Returns the names of servers that were
        successfully started.

        Args:
            bindings: List of :class:`LspBinding` objects from the
                actor compiler's :class:`CompilationMetadata`.
            workspace_path: Workspace root for all servers.

        Returns:
            Names of servers that were started (or already running).
        """
        started: list[str] = []
        for binding in bindings:
            server_name = getattr(binding, "lsp_server_name", "")
            if not server_name:
                continue
            try:
                self.start_server(server_name, workspace_path)
                started.append(server_name)
            except (LspError, LspServerNotFoundError) as exc:
                logger.warning(
                    "lsp.runtime.binding_start_failed",
                    server=server_name,
                    node=getattr(binding, "node_name", ""),
                    error=str(exc),
                )
        return started

    def deactivate_bindings(self, bindings: list[Any]) -> None:
        """Release LSP servers acquired by actor compilation bindings.

        Args:
            bindings: List of :class:`LspBinding` objects.
        """
        for binding in bindings:
            server_name = getattr(binding, "lsp_server_name", "")
            if not server_name:
                continue
            with contextlib.suppress(LspError, LspServerNotFoundError):
                self.stop_server(server_name)

    def stop_all(self) -> None:
        """Shut down all running LSP servers."""
        self._lifecycle.stop_all()


__all__ = [
    "LspRuntime",
]