python-sdk/pkg/hanzo-tools-lsp/hanzo_tools/lsp/lsp_tool.py

"""Language Server Protocol (LSP) tool for code intelligence.

One tool, two planes, chosen by the data the caller already supplies:

- no ``repo`` — a language server on the local tree, installed on demand and
  kept alive per ``language:root``. Answers what a working tree knows,
  including the edits it can apply back to disk.
- with ``repo`` — the indexed corpus behind ``/v1/code/lsp`` on api.hanzo.ai,
  which answers across a repo's dependencies without checking anything out.

``file`` names the file either way; ``repo`` says which world that file lives
in.
"""

import os
import json
import uuid
import atexit
import shutil
import asyncio
import logging
import tempfile
from typing import Any, Dict, List, Optional
from pathlib import Path
from dataclasses import field, dataclass
from urllib.parse import unquote, urlparse

from hanzo_tools.core import (
    NO_KEY,
    BaseTool,
    CloudError,
    HanzoCloud,
    MCPResourceDocument,
)

# LSP server configurations
LSP_SERVERS = {
    "go": {
        "name": "gopls",
        "install_cmd": ["go", "install", "golang.org/x/tools/gopls@latest"],
        "check_cmd": ["gopls", "version"],
        "start_cmd": ["gopls", "serve", "-mode=stdio"],
        "root_markers": ["go.work", "go.mod", "go.sum"],
        "file_extensions": [".go"],
        "capabilities": [
            "definition",
            "references",
            "rename",
            "diagnostics",
            "hover",
            "completion",
        ],
        "env": {"GOWORK": "auto"},
    },
    "python": {
        "name": "pyright",
        "install_cmd": ["npm", "install", "-g", "pyright"],
        "check_cmd": ["pyright-langserver", "--version"],
        "start_cmd": ["pyright-langserver", "--stdio"],
        "root_markers": [
            "pyproject.toml",
            "setup.py",
            "requirements.txt",
            "pyrightconfig.json",
        ],
        "file_extensions": [".py", ".pyi"],
        "capabilities": [
            "definition",
            "references",
            "rename",
            "diagnostics",
            "hover",
            "completion",
            "typeDefinition",
        ],
    },
    "typescript": {
        "name": "typescript-language-server",
        "install_cmd": [
            "npm",
            "install",
            "-g",
            "typescript",
            "typescript-language-server",
        ],
        "check_cmd": ["typescript-language-server", "--version"],
        "start_cmd": ["typescript-language-server", "--stdio"],
        "root_markers": ["tsconfig.json", "package.json"],
        "file_extensions": [".ts", ".tsx", ".js", ".jsx"],
        "capabilities": [
            "definition",
            "references",
            "rename",
            "diagnostics",
            "hover",
            "completion",
        ],
    },
    "rust": {
        "name": "rust-analyzer",
        "install_cmd": ["rustup", "component", "add", "rust-analyzer"],
        "check_cmd": ["rust-analyzer", "--version"],
        "start_cmd": ["rust-analyzer"],
        "root_markers": ["Cargo.toml"],
        "file_extensions": [".rs"],
        "capabilities": [
            "definition",
            "references",
            "rename",
            "diagnostics",
            "hover",
            "completion",
            "inlay_hints",
        ],
    },
    "java": {
        "name": "jdtls",
        "install_cmd": ["brew", "install", "jdtls"],
        "check_cmd": ["jdtls", "--version"],
        "start_cmd": ["jdtls"],
        "root_markers": ["pom.xml", "build.gradle", "build.gradle.kts"],
        "file_extensions": [".java"],
        "capabilities": [
            "definition",
            "references",
            "rename",
            "diagnostics",
            "hover",
            "completion",
        ],
    },
    "cpp": {
        "name": "clangd",
        "install_cmd": ["brew", "install", "llvm"],
        "check_cmd": ["clangd", "--version"],
        "start_cmd": ["clangd"],
        "root_markers": ["compile_commands.json", "CMakeLists.txt"],
        "file_extensions": [".cpp", ".cc", ".cxx", ".c", ".h", ".hpp"],
        "capabilities": [
            "definition",
            "references",
            "rename",
            "diagnostics",
            "hover",
            "completion",
        ],
    },
    "ruby": {
        "name": "solargraph",
        "install_cmd": ["gem", "install", "solargraph"],
        "check_cmd": ["solargraph", "--version"],
        "start_cmd": ["solargraph", "stdio"],
        "root_markers": ["Gemfile", ".solargraph.yml"],
        "file_extensions": [".rb"],
        "capabilities": [
            "definition",
            "references",
            "diagnostics",
            "hover",
            "completion",
        ],
    },
    "lua": {
        "name": "lua-language-server",
        "install_cmd": ["brew", "install", "lua-language-server"],
        "check_cmd": ["lua-language-server", "--version"],
        "start_cmd": ["lua-language-server"],
        "root_markers": [".luarc.json"],
        "file_extensions": [".lua"],
        "capabilities": [
            "definition",
            "references",
            "rename",
            "diagnostics",
            "hover",
            "completion",
        ],
    },
}


# Actions a local language server answers.
LOCAL_ACTIONS = [
    "definition",
    "references",
    "rename",
    "diagnostics",
    "hover",
    "completion",
    "code_action",
    "organize_imports",
    "status",
]

# The /v1/code/lsp op each action asks for, and the `relation` a locate op
# carries. "Where is X" is one question with four answers, so it is one op.
CLOUD_OPS: Dict[str, tuple] = {
    "hover": ("hover", None),
    "definition": ("locate", "definition"),
    "references": ("locate", "reference"),
    "type": ("locate", "type"),
    "implementation": ("locate", "implementation"),
    "symbols": ("symbols", None),
    "diagnostics": ("diagnostics", None),
    "completion": ("complete", None),
}


# Global LSP server registry - singleton per language:root_uri
_GLOBAL_SERVERS: Dict[str, "LSPServer"] = {}
_GLOBAL_LOCK = asyncio.Lock()
_CLEANUP_REGISTERED = False

logger = logging.getLogger(__name__)


def _cleanup_all_servers():
    """Cleanup all LSP servers on process exit."""
    for server in _GLOBAL_SERVERS.values():
        if server.process and server.process.returncode is None:
            try:
                server.process.terminate()
            except Exception:
                pass
    _GLOBAL_SERVERS.clear()


@dataclass
class LSPServer:
    """Represents an LSP server instance."""

    language: str
    process: Optional[asyncio.subprocess.Process]
    config: Dict[str, Any]
    root_uri: str
    initialized: bool = False
    request_id: int = field(default=0)
    pending_responses: Dict[int, asyncio.Future] = field(default_factory=dict)
    lock: asyncio.Lock = field(default_factory=asyncio.Lock)

    def next_id(self) -> int:
        """Get next request ID."""
        self.request_id += 1
        return self.request_id


class LSPTool(BaseTool):
    """Language Server Protocol tool for code intelligence."""

    name = "lsp"

    @property
    def description(self) -> str:
        return """Language Server Protocol tool for code intelligence.

    Local — a language server on your tree, installed as needed:
    - definition: Go to definition of symbol at position
    - references: Find all references to symbol
    - rename: Rename symbol across codebase
    - diagnostics: Get errors and warnings for file
    - hover: Get hover information at position
    - completion: Get code completions at position
    - code_action: Run LSP code actions
    - organize_imports: Organize imports for a file
    - status: Check LSP server status
    Supported languages: Go, Python, TypeScript/JavaScript, Rust, Java, C/C++, Ruby, Lua

    Cloud — pass `repo` (a git.hanzo.ai slug, optionally `rev`) to answer from
    the indexed corpus instead, which reaches across the repo's dependencies:
    definition, references, type, implementation, symbols, hover, completion,
    diagnostics. `file` is then the path within the repo.
    """

    def __init__(self):
        super().__init__()
        self._cloud: Optional[HanzoCloud] = None
        global _CLEANUP_REGISTERED
        if not _CLEANUP_REGISTERED:
            atexit.register(_cleanup_all_servers)
            _CLEANUP_REGISTERED = True

    def _get_language_from_file(self, file_path: str) -> Optional[str]:
        """Detect language from file extension."""
        ext = Path(file_path).suffix.lower()
        for lang, config in LSP_SERVERS.items():
            if ext in config["file_extensions"]:
                return lang
        return None

    def _find_project_root(self, file_path: str, language: str) -> str:
        """Find project root based on language markers."""
        path = Path(file_path).resolve()
        if language == "go":
            try:
                return str(self._find_go_workspace_root(path))
            except FileNotFoundError:
                return str(path.parent)
        markers = LSP_SERVERS[language]["root_markers"]
        for parent in path.parents:
            for marker in markers:
                if (parent / marker).is_file():
                    return str(parent)
        return str(path.parent)

    async def _check_lsp_installed(self, language: str) -> bool:
        """Check if LSP server is installed."""
        config = LSP_SERVERS.get(language)
        if not config:
            return False
        try:
            result = await asyncio.create_subprocess_exec(
                *config["check_cmd"],
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
            await result.communicate()
            return result.returncode == 0
        except (FileNotFoundError, OSError):
            return False

    async def _install_lsp(self, language: str) -> bool:
        """Install LSP server for language."""
        config = LSP_SERVERS.get(language)
        if not config:
            return False
        logger.info(f"Installing {config['name']} for {language}")
        try:
            installer = config["install_cmd"][0]
            if not shutil.which(installer):
                logger.error(f"Installer {installer} not found")
                return False
            result = await asyncio.create_subprocess_exec(
                *config["install_cmd"],
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
            stdout, stderr = await result.communicate()
            if result.returncode != 0:
                logger.error(f"Installation failed: {stderr.decode()}")
                return False
            logger.info(f"Successfully installed {config['name']}")
            return True
        except Exception as e:
            logger.error(f"Installation error: {e}")
            return False

    async def _ensure_lsp_running(
        self, language: str, root_uri: str
    ) -> Optional[LSPServer]:
        """Ensure LSP server is running for language."""
        server_key = f"{language}:{root_uri}"
        if server_key in _GLOBAL_SERVERS:
            server = _GLOBAL_SERVERS[server_key]
            if server.process and server.process.returncode is None:
                return server

        async with _GLOBAL_LOCK:
            if server_key in _GLOBAL_SERVERS:
                server = _GLOBAL_SERVERS[server_key]
                if server.process and server.process.returncode is None:
                    return server
                del _GLOBAL_SERVERS[server_key]

            if not await self._check_lsp_installed(language):
                if not await self._install_lsp(language):
                    return None

            config = LSP_SERVERS[language]
            try:
                env = os.environ.copy()
                env.update(config.get("env", {}))
                process = await asyncio.create_subprocess_exec(
                    *config["start_cmd"],
                    stdin=asyncio.subprocess.PIPE,
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE,
                    cwd=root_uri,
                    env=env,
                )
                server = LSPServer(
                    language=language, process=process, config=config, root_uri=root_uri
                )
                if not await self._initialize_lsp(server):
                    if server.process:
                        server.process.terminate()
                    return None
                _GLOBAL_SERVERS[server_key] = server
                logger.info(
                    f"Started global LSP server: {config['name']} for {root_uri}"
                )
                return server
            except Exception as e:
                logger.error(f"Failed to start LSP: {e}")
                return None

    async def _initialize_lsp(self, server: LSPServer) -> bool:
        """Send initialize request to LSP server."""
        root_uri = Path(server.root_uri).resolve().as_uri()
        init_params = {
            "processId": os.getpid(),
            "rootUri": root_uri,
            "rootPath": server.root_uri,
            "capabilities": {
                "workspace": {"workspaceFolders": True, "applyEdit": True},
                "textDocument": {
                    "synchronization": {"dynamicRegistration": True, "didSave": True},
                    "completion": {"completionItem": {"snippetSupport": True}},
                    "hover": {"contentFormat": ["markdown", "plaintext"]},
                    "definition": {"dynamicRegistration": True, "linkSupport": True},
                    "references": {"dynamicRegistration": True},
                    "rename": {"dynamicRegistration": True, "prepareSupport": True},
                },
            },
            "workspaceFolders": [{"uri": root_uri, "name": Path(server.root_uri).name}],
        }
        request = {
            "jsonrpc": "2.0",
            "id": server.next_id(),
            "method": "initialize",
            "params": init_params,
        }
        response = await self._send_request(server, request, timeout=60.0)
        if not response or "error" in response:
            logger.error(f"Failed to initialize LSP: {response}")
            return False
        await self._send_notification(server, "initialized", {})
        server.initialized = True
        return True

    async def _read_lsp_message(
        self, server: LSPServer, timeout: float = 30.0
    ) -> Optional[Dict[str, Any]]:
        """Read a single LSP message from server stdout."""
        if not server.process or not server.process.stdout:
            return None
        try:
            headers = {}
            while True:
                line = await asyncio.wait_for(
                    server.process.stdout.readline(), timeout=timeout
                )
                if not line:
                    return None
                line_str = line.decode("utf-8").strip()
                if not line_str:
                    break
                if ":" in line_str:
                    key, value = line_str.split(":", 1)
                    headers[key.strip().lower()] = value.strip()
            content_length = int(headers.get("content-length", 0))
            if content_length == 0:
                return None
            content = await asyncio.wait_for(
                server.process.stdout.read(content_length), timeout=timeout
            )
            return json.loads(content.decode("utf-8"))
        except asyncio.TimeoutError:
            return None
        except Exception as e:
            logger.error(f"Error reading LSP message: {e}")
            return None

    async def _send_request(
        self, server: LSPServer, request: Dict[str, Any], timeout: float = 30.0
    ) -> Optional[Dict[str, Any]]:
        """Send JSON-RPC request to LSP server and wait for response."""
        if (
            not server.process
            or server.process.returncode is not None
            or not server.process.stdin
        ):
            return None
        import time

        async with server.lock:
            try:
                request_str = json.dumps(request)
                content = request_str.encode("utf-8")
                header = f"Content-Length: {len(content)}\r\n\r\n"
                server.process.stdin.write(header.encode("utf-8") + content)
                await server.process.stdin.drain()
                request_id = request.get("id")
                start_time = time.monotonic()
                deadline = start_time + timeout
                while time.monotonic() < deadline:
                    remaining = max(0.1, deadline - time.monotonic())
                    response = await self._read_lsp_message(server, timeout=remaining)
                    if response is None:
                        if server.process.returncode is not None:
                            return None
                        continue
                    if "id" in response and response["id"] == request_id:
                        return response
                return None
            except Exception as e:
                logger.error(f"LSP communication error: {e}")
                return None

    async def _send_notification(
        self, server: LSPServer, method: str, params: Dict[str, Any]
    ) -> bool:
        """Send JSON-RPC notification (no response expected)."""
        if (
            not server.process
            or server.process.returncode is not None
            or not server.process.stdin
        ):
            return False
        try:
            notification = {"jsonrpc": "2.0", "method": method, "params": params}
            content = json.dumps(notification).encode("utf-8")
            header = f"Content-Length: {len(content)}\r\n\r\n"
            server.process.stdin.write(header.encode("utf-8") + content)
            await server.process.stdin.drain()
            return True
        except Exception as e:
            logger.error(f"Failed to send notification: {e}")
            return False

    async def run(
        self,
        action: str,
        file: Optional[str] = None,
        line: Optional[int] = None,
        character: Optional[int] = None,
        new_name: Optional[str] = None,
        apply_edits: bool = False,
        repo: Optional[str] = None,
        rev: Optional[str] = None,
        **kwargs,
    ) -> MCPResourceDocument:
        """Execute LSP action, locally or against the cloud index."""
        if not file or not file.strip():
            return MCPResourceDocument(
                data={
                    "error": "file is required: a local path, or the path within `repo`"
                }
            )

        if action not in LOCAL_ACTIONS and action not in CLOUD_OPS:
            known = LOCAL_ACTIONS + [a for a in CLOUD_OPS if a not in LOCAL_ACTIONS]
            return MCPResourceDocument(
                data={"error": f"Invalid action. Must be one of: {', '.join(known)}"}
            )

        if repo and repo.strip():
            return MCPResourceDocument(
                data=await self._cloud_action(
                    action, repo.strip(), file, line, character, rev
                )
            )

        if action not in LOCAL_ACTIONS:
            return MCPResourceDocument(
                data={
                    "error": f"Action '{action}' is answered by the cloud index; pass `repo`",
                    "local_actions": LOCAL_ACTIONS,
                }
            )

        language = self._get_language_from_file(file)
        if not language:
            return MCPResourceDocument(
                data={
                    "error": f"Unsupported file type: {file}",
                    "supported_languages": list(LSP_SERVERS.keys()),
                }
            )

        capabilities = LSP_SERVERS[language]["capabilities"]
        if action not in capabilities and action not in [
            "status",
            "organize_imports",
            "code_action",
        ]:
            return MCPResourceDocument(
                data={
                    "error": f"Action '{action}' not supported for {language}",
                    "supported_actions": capabilities,
                }
            )

        if action == "status":
            installed = await self._check_lsp_installed(language)
            return MCPResourceDocument(
                data={
                    "language": language,
                    "lsp_server": LSP_SERVERS[language]["name"],
                    "installed": installed,
                    "capabilities": capabilities,
                }
            )

        root_uri = self._find_project_root(file, language)
        server = await self._ensure_lsp_running(language, root_uri)
        if not server:
            return MCPResourceDocument(
                data={
                    "error": f"Failed to start LSP server for {language}",
                    "install_command": " ".join(LSP_SERVERS[language]["install_cmd"]),
                }
            )

        result = await self._execute_lsp_action(
            server,
            action,
            file,
            line,
            character,
            new_name,
            apply_edits,
            range_spec=kwargs.get("range"),
            only=kwargs.get("only"),
        )
        return MCPResourceDocument(data=result)

    async def _cloud_action(
        self,
        action: str,
        repo: str,
        file: str,
        line: Optional[int],
        character: Optional[int],
        rev: Optional[str],
    ) -> Dict[str, Any]:
        """Ask the cloud index. The answer is whatever /v1/code/lsp returns."""
        op, relation = CLOUD_OPS.get(action, (None, None))
        if op is None:
            return {
                "action": action,
                "repo": repo,
                "error": f"Action '{action}' needs a working tree; call it without `repo`",
                "cloud_actions": list(CLOUD_OPS),
            }

        if self._cloud is None:
            self._cloud = HanzoCloud()
        if not self._cloud.configured():
            return {
                "action": action,
                "repo": repo,
                "error": NO_KEY,
            }

        # The wire speaks LSP's own frame: 0-based line, 0-based UTF-16
        # character. The tool's `line` is 1-based, so it is shifted here exactly
        # as the local plane shifts it before a JSON-RPC request.
        body: Dict[str, Any] = {
            "repo": repo,
            "path": file,
            "line": (line - 1) if line else 0,
            "character": character if character else 0,
        }
        if rev and rev.strip():
            body["rev"] = rev.strip()
        if relation:
            body["relation"] = relation

        try:
            return await self._cloud.post(f"/v1/code/lsp/{op}", body)
        except CloudError as e:
            return {"action": action, "repo": repo, "error": str(e)}

    async def call(self, **kwargs) -> str:
        """Tool interface for MCP - converts result to JSON string."""
        result = await self.run(**kwargs)
        return result.to_json_string()

    def register(self, mcp_server) -> None:
        """Register tool with MCP server."""

        @mcp_server.tool(name=self.name, description=self.description)
        async def lsp_handler(
            action: str,
            file: str,
            line: Optional[int] = None,
            character: Optional[int] = None,
            new_name: Optional[str] = None,
            apply_edits: bool = False,
            only: Optional[List[str]] = None,
            range: Optional[Dict[str, Dict[str, int]]] = None,
            repo: Optional[str] = None,
            rev: Optional[str] = None,
        ) -> str:
            return await self.call(
                action=action,
                file=file,
                line=line,
                character=character,
                new_name=new_name,
                apply_edits=apply_edits,
                only=only,
                range=range,
                repo=repo,
                rev=rev,
            )

    def _path_to_uri(self, path: str) -> str:
        return Path(path).resolve().as_uri()

    def _uri_to_path(self, uri: str) -> str:
        if uri.startswith("file://"):
            parsed = urlparse(uri)
            path = unquote(parsed.path)
            if (
                path.startswith("/")
                and len(path) >= 3
                and path[2] == ":"
                and path[1].isalpha()
            ):
                path = path[1:]
            return path
        return uri

    def _find_go_workspace_root(self, start: Path) -> Path:
        p = start.resolve()
        if p.is_file():
            p = p.parent
        for d in (p, *p.parents):
            work = d / "go.work"
            if work.is_file():
                return d
        for d in (p, *p.parents):
            mod = d / "go.mod"
            if mod.is_file():
                return d
        raise FileNotFoundError(f"no go.work or go.mod found above {start}")

    def _is_within_root(self, path: str, root_dir: str) -> bool:
        try:
            Path(path).resolve().relative_to(Path(root_dir).resolve())
            return True
        except Exception:
            return False

    async def _open_document(self, server: LSPServer, file_path: str) -> bool:
        """Notify LSP server that a document is opened."""
        abs_path = str(Path(file_path).resolve())
        try:
            with open(abs_path, "r", encoding="utf-8") as f:
                content = f.read()
        except Exception as e:
            logger.error(f"Failed to read file {file_path}: {e}")
            return False
        language_id = server.language
        if server.language == "typescript":
            ext = Path(file_path).suffix.lower()
            if ext in [".tsx", ".jsx"]:
                language_id = "typescriptreact" if "ts" in ext else "javascriptreact"
            elif ext in [".js", ".mjs", ".cjs"]:
                language_id = "javascript"
        params = {
            "textDocument": {
                "uri": self._path_to_uri(abs_path),
                "languageId": language_id,
                "version": 1,
                "text": content,
            }
        }
        return await self._send_notification(server, "textDocument/didOpen", params)

    def _parse_location(self, location: Dict[str, Any]) -> Dict[str, Any]:
        """Parse LSP Location into a readable format."""
        uri = location.get("uri", "")
        file_path = self._uri_to_path(uri)
        range_info = location.get("range", {})
        start = range_info.get("start", {})
        end = range_info.get("end", {})
        return {
            "file": file_path,
            "start": {
                "line": start.get("line", 0) + 1,
                "character": start.get("character", 0),
            },
            "end": {
                "line": end.get("line", 0) + 1,
                "character": end.get("character", 0),
            },
        }

    def _utf16_index_to_py_index(self, text: str, utf16_index: int) -> int:
        if utf16_index <= 0:
            return 0
        units = 0
        for idx, ch in enumerate(text):
            units += 2 if ord(ch) > 0xFFFF else 1
            if units >= utf16_index:
                return idx + 1
        return len(text)

    def _utf16_len(self, text: str) -> int:
        units = 0
        for ch in text:
            units += 2 if ord(ch) > 0xFFFF else 1
        return units

    def _lsp_position_to_offset(
        self, lines: List[str], line: int, character: int
    ) -> int:
        if line < 0:
            return 0
        if line >= len(lines):
            return sum(len(l) for l in lines)
        offset = sum(len(l) for l in lines[:line])
        line_text = lines[line]
        return offset + self._utf16_index_to_py_index(line_text, character)

    def _file_range_for_code_action(self, file_path: str) -> Dict[str, Dict[str, int]]:
        try:
            with open(file_path, "r", encoding="utf-8") as f:
                content = f.read()
        except Exception:
            return {
                "start": {"line": 0, "character": 0},
                "end": {"line": 0, "character": 0},
            }
        if not content:
            return {
                "start": {"line": 0, "character": 0},
                "end": {"line": 0, "character": 0},
            }
        lines = content.splitlines()
        last_line_index = len(lines) - 1
        end_char = self._utf16_len(lines[last_line_index])
        return {
            "start": {"line": 0, "character": 0},
            "end": {"line": last_line_index, "character": end_char},
        }

    def _apply_text_edits(self, file_path: str, edits: List[Dict[str, Any]]) -> None:
        with open(file_path, "r", encoding="utf-8") as f:
            content = f.read()
        content = self._render_text_edits(content, edits)
        with open(file_path, "w", encoding="utf-8") as f:
            f.write(content)

    def _render_text_edits(self, content: str, edits: List[Dict[str, Any]]) -> str:
        lines = content.splitlines(keepends=True)
        normalized = []
        for edit in edits:
            range_info = edit.get("range", {})
            start = range_info.get("start", {})
            end = range_info.get("end", {})
            start_offset = self._lsp_position_to_offset(
                lines, start.get("line", 0), start.get("character", 0)
            )
            end_offset = self._lsp_position_to_offset(
                lines, end.get("line", 0), end.get("character", 0)
            )
            normalized.append(
                (
                    start.get("line", 0),
                    start.get("character", 0),
                    start_offset,
                    end_offset,
                    edit.get("newText", ""),
                )
            )
        normalized.sort(key=lambda item: (item[0], item[1]), reverse=True)
        for _, _, start_offset, end_offset, new_text in normalized:
            content = content[:start_offset] + new_text + content[end_offset:]
        return content

    def _workspace_edit_files(self, edit: Dict[str, Any]) -> List[str]:
        files: List[str] = []
        if "documentChanges" in edit:
            for change in edit["documentChanges"]:
                if "kind" in change:
                    if change.get("kind") == "rename":
                        files.append(self._uri_to_path(change.get("oldUri", "")))
                        files.append(self._uri_to_path(change.get("newUri", "")))
                    elif change.get("kind") in ["create", "delete"]:
                        files.append(self._uri_to_path(change.get("uri", "")))
                elif "textDocument" in change:
                    files.append(self._uri_to_path(change["textDocument"]["uri"]))
        elif "changes" in edit:
            for uri in edit["changes"].keys():
                files.append(self._uri_to_path(uri))
        return sorted({f for f in files if f})

    def _apply_workspace_edit(
        self, edit: Dict[str, Any], root_dir: str
    ) -> tuple[list[str], list[str]]:
        applied: list[str] = []
        errors: list[str] = []
        root_dir = str(Path(root_dir).resolve())

        changes = edit.get("documentChanges")
        if changes is None:
            changes = [
                {"textDocument": {"uri": uri}, "edits": edits}
                for uri, edits in edit.get("changes", {}).items()
            ]

        if not changes:
            return applied, errors

        backup_dir = Path(tempfile.mkdtemp(prefix="hanzo-lsp-", dir=root_dir))
        backups: Dict[str, str] = {}
        mtimes: Dict[str, float] = {}
        created: set[str] = set()

        def backup_path(path: str) -> str:
            rel = os.path.relpath(path, root_dir)
            dst = backup_dir / rel
            dst.parent.mkdir(parents=True, exist_ok=True)
            return str(dst)

        try:
            # Preflight: collect backups
            for change in changes:
                if "kind" in change:
                    kind = change.get("kind")
                    if kind == "rename":
                        old_path = self._uri_to_path(change.get("oldUri", ""))
                        new_path = self._uri_to_path(change.get("newUri", ""))
                        if not self._is_within_root(
                            old_path, root_dir
                        ) or not self._is_within_root(new_path, root_dir):
                            raise RuntimeError(
                                f"rename outside workspace root: {old_path} -> {new_path}"
                            )
                        if os.path.exists(old_path) and old_path not in backups:
                            backups[old_path] = backup_path(old_path)
                            shutil.copy2(old_path, backups[old_path])
                            mtimes[old_path] = os.path.getmtime(old_path)
                        if os.path.exists(new_path) and new_path not in backups:
                            backups[new_path] = backup_path(new_path)
                            shutil.copy2(new_path, backups[new_path])
                            mtimes[new_path] = os.path.getmtime(new_path)
                    elif kind == "create":
                        file_path = self._uri_to_path(change.get("uri", ""))
                        if not self._is_within_root(file_path, root_dir):
                            raise RuntimeError(
                                f"create outside workspace root: {file_path}"
                            )
                        if os.path.exists(file_path) and file_path not in backups:
                            backups[file_path] = backup_path(file_path)
                            shutil.copy2(file_path, backups[file_path])
                            mtimes[file_path] = os.path.getmtime(file_path)
                    elif kind == "delete":
                        file_path = self._uri_to_path(change.get("uri", ""))
                        if not self._is_within_root(file_path, root_dir):
                            raise RuntimeError(
                                f"delete outside workspace root: {file_path}"
                            )
                        if os.path.exists(file_path) and file_path not in backups:
                            backups[file_path] = backup_path(file_path)
                            shutil.copy2(file_path, backups[file_path])
                            mtimes[file_path] = os.path.getmtime(file_path)
                elif "textDocument" in change and "edits" in change:
                    file_path = self._uri_to_path(change["textDocument"]["uri"])
                    if not self._is_within_root(file_path, root_dir):
                        raise RuntimeError(f"edit outside workspace root: {file_path}")
                    if os.path.exists(file_path) and file_path not in backups:
                        backups[file_path] = backup_path(file_path)
                        shutil.copy2(file_path, backups[file_path])
                        mtimes[file_path] = os.path.getmtime(file_path)

            # Apply changes in order
            for change in changes:
                if "kind" in change:
                    kind = change.get("kind")
                    if kind == "rename":
                        old_path = self._uri_to_path(change.get("oldUri", ""))
                        new_path = self._uri_to_path(change.get("newUri", ""))
                        options = change.get("options", {})
                        if os.path.exists(new_path):
                            if options.get("ignoreIfExists"):
                                continue
                            if not options.get("overwrite"):
                                raise RuntimeError(f"rename target exists: {new_path}")
                        Path(new_path).parent.mkdir(parents=True, exist_ok=True)
                        shutil.move(old_path, new_path)
                        applied.append(new_path)
                        continue
                    if kind == "create":
                        file_path = self._uri_to_path(change.get("uri", ""))
                        options = change.get("options", {})
                        if os.path.exists(file_path):
                            if options.get("ignoreIfExists"):
                                continue
                            if not options.get("overwrite"):
                                raise RuntimeError(f"create target exists: {file_path}")
                        Path(file_path).parent.mkdir(parents=True, exist_ok=True)
                        with open(file_path, "w", encoding="utf-8") as f:
                            f.write(change.get("content", ""))
                        created.add(file_path)
                        applied.append(file_path)
                        continue
                    if kind == "delete":
                        file_path = self._uri_to_path(change.get("uri", ""))
                        options = change.get("options", {})
                        if not os.path.exists(file_path):
                            if options.get("ignoreIfNotExists"):
                                continue
                            raise RuntimeError(f"delete target missing: {file_path}")
                        if os.path.isdir(file_path):
                            if not options.get("recursive"):
                                raise RuntimeError(
                                    f"delete target is directory: {file_path}"
                                )
                            shutil.rmtree(file_path)
                        else:
                            os.unlink(file_path)
                        applied.append(file_path)
                        continue
                    raise RuntimeError(f"unsupported documentChange: {kind}")

                if "textDocument" in change and "edits" in change:
                    file_path = self._uri_to_path(change["textDocument"]["uri"])
                    if file_path in mtimes and os.path.exists(file_path):
                        if os.path.getmtime(file_path) != mtimes[file_path]:
                            raise RuntimeError(f"conflict detected for {file_path}")
                    with open(file_path, "r", encoding="utf-8") as f:
                        content = f.read()
                    updated = self._render_text_edits(content, change["edits"])
                    temp_path = f"{file_path}.hanzo_tmp_{uuid.uuid4().hex}"
                    with open(temp_path, "w", encoding="utf-8") as f:
                        f.write(updated)
                    os.replace(temp_path, file_path)
                    applied.append(file_path)
                else:
                    raise RuntimeError(
                        f"unsupported documentChange: {change.get('kind', 'unknown')}"
                    )

        except Exception as exc:
            errors.append(str(exc))
            # Rollback
            for path, backup in backups.items():
                try:
                    Path(path).parent.mkdir(parents=True, exist_ok=True)
                    shutil.copy2(backup, path)
                except Exception:
                    pass
            for path in created:
                if path not in backups:
                    try:
                        if os.path.isdir(path):
                            shutil.rmtree(path)
                        else:
                            os.unlink(path)
                    except Exception:
                        pass
            applied = []
        finally:
            shutil.rmtree(backup_dir, ignore_errors=True)

        return applied, errors

    async def _execute_lsp_action(
        self,
        server: LSPServer,
        action: str,
        file: str,
        line: Optional[int],
        character: Optional[int],
        new_name: Optional[str],
        apply_edits: bool,
        range_spec: Optional[Dict[str, Any]] = None,
        only: Optional[List[str]] = None,
    ) -> Dict[str, Any]:
        """Execute specific LSP action."""
        abs_path = str(Path(file).resolve())
        uri = self._path_to_uri(abs_path)
        await self._open_document(server, file)
        position = {
            "line": (line - 1) if line else 0,
            "character": character if character else 0,
        }

        if action == "definition":
            request = {
                "jsonrpc": "2.0",
                "id": server.next_id(),
                "method": "textDocument/definition",
                "params": {"textDocument": {"uri": uri}, "position": position},
            }
            response = await self._send_request(server, request)
            if response and "result" in response:
                result = response["result"]
                if result is None:
                    return {
                        "action": "definition",
                        "file": file,
                        "result": None,
                        "message": "No definition found",
                    }
                if isinstance(result, list):
                    return {
                        "action": "definition",
                        "file": file,
                        "definitions": [self._parse_location(loc) for loc in result],
                    }
                elif isinstance(result, dict):
                    return {
                        "action": "definition",
                        "file": file,
                        "definition": self._parse_location(result),
                    }
            return {
                "action": "definition",
                "file": file,
                "error": response.get("error") if response else "No response",
            }

        elif action == "references":
            request = {
                "jsonrpc": "2.0",
                "id": server.next_id(),
                "method": "textDocument/references",
                "params": {
                    "textDocument": {"uri": uri},
                    "position": position,
                    "context": {"includeDeclaration": True},
                },
            }
            response = await self._send_request(server, request)
            if response and "result" in response:
                result = response["result"]
                refs = (
                    [self._parse_location(loc) for loc in result]
                    if isinstance(result, list)
                    else []
                )
                return {
                    "action": "references",
                    "file": file,
                    "references": refs,
                    "count": len(refs),
                }
            return {
                "action": "references",
                "file": file,
                "error": response.get("error") if response else "No response",
            }

        elif action == "rename":
            if not new_name:
                return {
                    "action": "rename",
                    "error": "new_name is required for rename action",
                }
            request = {
                "jsonrpc": "2.0",
                "id": server.next_id(),
                "method": "textDocument/rename",
                "params": {
                    "textDocument": {"uri": uri},
                    "position": position,
                    "newName": new_name,
                },
            }
            response = await self._send_request(server, request)
            if response and "result" in response:
                result = response["result"]
                if result is None:
                    return {
                        "action": "rename",
                        "file": file,
                        "error": "Rename not possible at this location",
                    }
                changes = {}
                if "changes" in result:
                    for file_uri, edits in result["changes"].items():
                        file_path = self._uri_to_path(file_uri)
                        changes[file_path] = [
                            {
                                "range": self._parse_location(
                                    {"uri": file_uri, "range": edit["range"]}
                                ),
                                "newText": edit["newText"],
                            }
                            for edit in edits
                        ]
                applied = []
                apply_errors: List[str] = []
                touched_files = self._workspace_edit_files(result)
                if apply_edits:
                    applied, apply_errors = self._apply_workspace_edit(
                        result, server.root_uri
                    )
                payload: Dict[str, Any] = {
                    "action": "rename",
                    "file": file,
                    "new_name": new_name,
                    "changes": changes,
                    "files_affected": len(changes),
                    "touched_files": touched_files,
                }
                if apply_edits:
                    payload["applied_files"] = applied
                    if apply_errors:
                        payload["apply_errors"] = apply_errors
                return payload
            return {
                "action": "rename",
                "file": file,
                "error": response.get("error") if response else "No response",
            }

        elif action == "hover":
            request = {
                "jsonrpc": "2.0",
                "id": server.next_id(),
                "method": "textDocument/hover",
                "params": {"textDocument": {"uri": uri}, "position": position},
            }
            response = await self._send_request(server, request)
            if response and "result" in response:
                result = response["result"]
                if result is None:
                    return {
                        "action": "hover",
                        "file": file,
                        "result": None,
                        "message": "No hover info",
                    }
                contents = result.get("contents", "")
                if isinstance(contents, dict):
                    hover_text = contents.get("value", str(contents))
                elif isinstance(contents, list):
                    hover_text = "\n".join(
                        c.get("value", str(c)) if isinstance(c, dict) else str(c)
                        for c in contents
                    )
                else:
                    hover_text = str(contents)
                return {
                    "action": "hover",
                    "file": file,
                    "position": {"line": line, "character": character},
                    "contents": hover_text,
                }
            return {
                "action": "hover",
                "file": file,
                "error": response.get("error") if response else "No response",
            }

        elif action == "completion":
            request = {
                "jsonrpc": "2.0",
                "id": server.next_id(),
                "method": "textDocument/completion",
                "params": {"textDocument": {"uri": uri}, "position": position},
            }
            response = await self._send_request(server, request, timeout=10.0)
            if response and "result" in response:
                result = response["result"]
                if result is None:
                    return {
                        "action": "completion",
                        "file": file,
                        "completions": [],
                        "count": 0,
                    }
                items = result if isinstance(result, list) else result.get("items", [])
                completions = [
                    {
                        "label": item.get("label", ""),
                        "kind": item.get("kind", 0),
                        "detail": item.get("detail", ""),
                    }
                    for item in items[:50]
                ]
                return {
                    "action": "completion",
                    "file": file,
                    "position": {"line": line, "character": character},
                    "completions": completions,
                    "count": len(completions),
                }
            return {
                "action": "completion",
                "file": file,
                "error": response.get("error") if response else "No response",
            }

        elif action == "code_action":
            action_range = range_spec or self._file_range_for_code_action(abs_path)
            context: Dict[str, Any] = {}
            if only:
                context["only"] = only
            request = {
                "jsonrpc": "2.0",
                "id": server.next_id(),
                "method": "textDocument/codeAction",
                "params": {
                    "textDocument": {"uri": uri},
                    "range": action_range,
                    "context": context,
                },
            }
            response = await self._send_request(server, request)
            if response and "result" in response:
                actions = response["result"] or []
                edits_applied: list[str] = []
                apply_errors: list[str] = []
                touched_files: list[str] = []
                commands: list[Dict[str, Any]] = []
                edits_seen = 0
                for action_item in actions:
                    if "edit" in action_item:
                        edits_seen += 1
                        touched_files.extend(
                            self._workspace_edit_files(action_item["edit"])
                        )
                        if apply_edits:
                            applied, errors = self._apply_workspace_edit(
                                action_item["edit"], server.root_uri
                            )
                            edits_applied.extend(applied)
                            apply_errors.extend(errors)
                    if "command" in action_item:
                        commands.append(action_item["command"])
                payload: Dict[str, Any] = {
                    "action": "code_action",
                    "file": file,
                    "edits_found": edits_seen,
                    "touched_files": sorted(set(touched_files)),
                    "commands": commands,
                }
                if apply_edits:
                    payload["applied_files"] = edits_applied
                    if apply_errors:
                        payload["apply_errors"] = apply_errors
                return payload
            return {
                "action": "code_action",
                "file": file,
                "error": response.get("error") if response else "No response",
            }

        elif action == "organize_imports":
            file_range = self._file_range_for_code_action(abs_path)
            request = {
                "jsonrpc": "2.0",
                "id": server.next_id(),
                "method": "textDocument/codeAction",
                "params": {
                    "textDocument": {"uri": uri},
                    "range": file_range,
                    "context": {"only": ["source.organizeImports"]},
                },
            }
            response = await self._send_request(server, request)
            if response and "result" in response:
                actions = response["result"] or []
                edits_applied: list[str] = []
                apply_errors: list[str] = []
                edits_seen = 0
                touched_files: list[str] = []
                for action_item in actions:
                    kind = action_item.get("kind")
                    if kind and kind != "source.organizeImports":
                        continue
                    if "edit" in action_item:
                        edits_seen += 1
                        touched_files.extend(
                            self._workspace_edit_files(action_item["edit"])
                        )
                        if apply_edits:
                            applied, errors = self._apply_workspace_edit(
                                action_item["edit"], server.root_uri
                            )
                            edits_applied.extend(applied)
                            apply_errors.extend(errors)
                payload: Dict[str, Any] = {
                    "action": "organize_imports",
                    "file": file,
                    "edits_found": edits_seen,
                    "touched_files": sorted(set(touched_files)),
                }
                if apply_edits:
                    payload["applied_files"] = edits_applied
                    if apply_errors:
                        payload["apply_errors"] = apply_errors
                return payload
            return {
                "action": "organize_imports",
                "file": file,
                "error": response.get("error") if response else "No response",
            }

        elif action == "diagnostics":
            return {
                "action": "diagnostics",
                "file": file,
                "note": "Diagnostics are push-based; use language-specific tools (go vet, pylint, etc.) for on-demand checking",
            }

        return {"error": f"Unknown action: {action}"}

    async def cleanup(self):
        """Clean up LSP servers."""
        async with _GLOBAL_LOCK:
            for server in list(_GLOBAL_SERVERS.values()):
                if server.process and server.process.returncode is None:
                    try:
                        server.process.terminate()
                        await asyncio.wait_for(server.process.wait(), timeout=5.0)
                    except Exception:
                        pass
            _GLOBAL_SERVERS.clear()


def create_lsp_tool():
    """Factory function to create LSP tool."""
    return LSPTool()