clangd_client.py

"""
Clangd LSP client — manages a clangd subprocess for precision C++ queries.

Handles the full LSP lifecycle: initialize, didOpen, requests, shutdown.
Uses a reader thread for async JSON-RPC message consumption.
"""

import json
import os
import subprocess
import sys
import threading
import time
from pathlib import Path
from typing import Optional
from urllib.parse import quote as url_quote


class ClangdClient:
    """Manages a clangd subprocess and provides LSP request methods."""

    def __init__(
        self,
        clangd_path: str = "clangd",
        compile_commands_dir: str = "",
        project_root: str = "",
    ):
        self.clangd_path = clangd_path
        self.compile_commands_dir = compile_commands_dir
        self.project_root = Path(project_root)

        self._process: Optional[subprocess.Popen] = None
        self._reader_thread: Optional[threading.Thread] = None
        self._stderr_thread: Optional[threading.Thread] = None
        self._request_id = 0
        self._pending: dict[int, threading.Event] = {}
        self._responses: dict[int, dict] = {}
        self._lock = threading.Lock()
        self._write_lock = threading.Lock()
        self._start_lock = threading.Lock()
        self._open_files: set[str] = set()
        self._open_files_lock = threading.Lock()
        self._initialized = False
        self._alive = False

    @property
    def available(self) -> bool:
        """Check if clangd binary exists."""
        return Path(self.clangd_path).exists()

    def ensure_started(self) -> bool:
        """Start clangd if not running. Returns True if ready."""
        with self._start_lock:
            if self._alive and self._process and self._process.poll() is None:
                return True

            if not self.available:
                print(f"[code-intel] clangd not found at {self.clangd_path}", file=sys.stderr)
                return False

            try:
                self._start()
                return True
            except Exception as e:
                print(f"[code-intel] Failed to start clangd: {e}", file=sys.stderr)
                return False

    def _start(self) -> None:
        """Launch clangd subprocess and initialize LSP."""
        cmd = [
            self.clangd_path,
            f"--compile-commands-dir={self.compile_commands_dir}",
            "--background-index",
            "-j=4",
            "--pch-storage=memory",
            "--header-insertion=never",
        ]
        print(f"[code-intel] Starting clangd: {' '.join(cmd)}", file=sys.stderr)

        self._process = subprocess.Popen(
            cmd,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

        self._alive = True
        self._reader_thread = threading.Thread(target=self._reader_loop, daemon=True)
        self._reader_thread.start()

        # Also start stderr reader to prevent buffer fill
        self._stderr_thread = threading.Thread(target=self._stderr_loop, daemon=True)
        self._stderr_thread.start()

        # Send initialize
        root_uri = self._path_to_uri(str(self.project_root))
        resp = self._send_request("initialize", {
            "processId": os.getpid(),
            "rootUri": root_uri,
            "capabilities": {
                "textDocument": {
                    "hover": {"contentFormat": ["plaintext", "markdown"]},
                    "definition": {},
                    "references": {},
                    "typeHierarchy": {},
                    "callHierarchy": {},
                },
            },
            "initializationOptions": {},
        })

        if resp is None:
            raise RuntimeError("clangd initialize timed out")

        # Send initialized notification
        self._send_notification("initialized", {})
        self._initialized = True
        print("[code-intel] clangd initialized", file=sys.stderr)

    def _reader_loop(self) -> None:
        """Read JSON-RPC messages from clangd stdout."""
        try:
            while self._alive and self._process and self._process.poll() is None:
                # Read headers line by line until blank line
                headers: dict[str, str] = {}
                while self._alive:
                    raw_line = self._process.stdout.readline()
                    if not raw_line:
                        self._alive = False
                        self._wake_all_pending()
                        return
                    line = raw_line.decode("utf-8", errors="replace").strip()
                    if not line:
                        break  # blank line = end of headers
                    if ":" in line:
                        key, _, value = line.partition(":")
                        headers[key.strip().lower()] = value.strip()

                content_length = int(headers.get("content-length", "0"))
                if content_length == 0:
                    continue

                body = b""
                while len(body) < content_length:
                    chunk = self._process.stdout.read(content_length - len(body))
                    if not chunk:
                        self._alive = False
                        self._wake_all_pending()
                        return
                    body += chunk

                try:
                    msg = json.loads(body.decode("utf-8"))
                except json.JSONDecodeError:
                    continue

                # Dispatch response
                if "id" in msg and ("result" in msg or "error" in msg):
                    msg_id = msg["id"]
                    with self._lock:
                        self._responses[msg_id] = msg
                        event = self._pending.get(msg_id)
                    if event:
                        event.set()

        except Exception as e:
            print(f"[code-intel] Reader thread error: {e}", file=sys.stderr)
            self._alive = False
            self._wake_all_pending()

    def _wake_all_pending(self) -> None:
        """Wake all blocked request threads (e.g., when clangd crashes)."""
        with self._lock:
            for event in self._pending.values():
                event.set()

    def _stderr_loop(self) -> None:
        """Drain clangd stderr to prevent buffer fill."""
        try:
            while self._alive and self._process and self._process.poll() is None:
                line = self._process.stderr.readline()
                if not line:
                    break
        except Exception:
            pass

    def _send_request(self, method: str, params: dict, timeout: float = 30.0) -> Optional[dict]:
        """Send a JSON-RPC request and wait for response."""
        with self._lock:
            self._request_id += 1
            req_id = self._request_id
            event = threading.Event()
            self._pending[req_id] = event

        msg = {
            "jsonrpc": "2.0",
            "id": req_id,
            "method": method,
            "params": params,
        }
        self._write_message(msg)

        if not event.wait(timeout=timeout):
            print(f"[code-intel] Request {method} (id={req_id}) timed out after {timeout}s", file=sys.stderr)
            with self._lock:
                self._pending.pop(req_id, None)
                self._responses.pop(req_id, None)  # clean up any late arrival
            return None

        with self._lock:
            self._pending.pop(req_id, None)
            resp = self._responses.pop(req_id, None)

        if resp and "error" in resp:
            print(f"[code-intel] LSP error for {method}: {resp['error']}", file=sys.stderr)
            return None

        return resp.get("result") if resp else None

    def _send_notification(self, method: str, params: dict) -> None:
        """Send a JSON-RPC notification (no response expected)."""
        msg = {
            "jsonrpc": "2.0",
            "method": method,
            "params": params,
        }
        self._write_message(msg)

    def _write_message(self, msg: dict) -> None:
        """Write a JSON-RPC message to clangd stdin. Thread-safe."""
        body = json.dumps(msg).encode("utf-8")
        header = f"Content-Length: {len(body)}\r\n\r\n".encode("ascii")
        with self._write_lock:
            try:
                if not self._process or not self._process.stdin:
                    self._alive = False
                    return
                self._process.stdin.write(header + body)
                self._process.stdin.flush()
            except (BrokenPipeError, OSError) as e:
                print(f"[code-intel] Write failed: {e}", file=sys.stderr)
                self._alive = False

    def _path_to_uri(self, path: str) -> str:
        """Convert a file path to a file:// URI."""
        path = path.replace("\\", "/")
        if not path.startswith("/"):
            path = "/" + path
        # Encode special characters but keep / and :
        encoded = url_quote(path, safe="/:")
        return f"file://{encoded}"

    def _uri_to_relative_path(self, uri: str) -> str:
        """Convert a file:// URI back to a project-relative path."""
        path = uri.replace("file:///", "").replace("file://", "")
        path = path.replace("\\", "/")
        # Strip project root prefix to get relative path
        root = str(self.project_root).replace("\\", "/")
        if not root.endswith("/"):
            root += "/"
        if path.startswith(root):
            path = path[len(root):]
        return path

    def _resolve_path(self, file: str) -> str:
        """Resolve a potentially relative path to absolute."""
        p = Path(file)
        if not p.is_absolute():
            p = self.project_root / file
        return str(p).replace("\\", "/")

    def ensure_file_open(self, file: str) -> str:
        """Send didOpen for a file if not already open. Returns the URI."""
        abs_path = self._resolve_path(file)
        uri = self._path_to_uri(abs_path)

        with self._open_files_lock:
            if uri in self._open_files:
                return uri
            # Mark as open before sending to prevent duplicate didOpen
            self._open_files.add(uri)

        try:
            content = Path(abs_path).read_text(encoding="utf-8", errors="replace")
        except FileNotFoundError:
            with self._open_files_lock:
                self._open_files.discard(uri)
            raise FileNotFoundError(f"File not found: {abs_path}")

        # Determine language ID
        lang = "cpp"
        if abs_path.endswith(".c"):
            lang = "c"

        self._send_notification("textDocument/didOpen", {
            "textDocument": {
                "uri": uri,
                "languageId": lang,
                "version": 1,
                "text": content,
            }
        })
        # Small delay to let clangd process the file
        time.sleep(0.1)
        return uri

    def shutdown(self) -> None:
        """Gracefully shut down clangd."""
        if self._process and self._process.poll() is None:
            try:
                self._send_request("shutdown", {}, timeout=5.0)
                self._send_notification("exit", {})
            except Exception:
                pass
            self._alive = False
            try:
                self._process.terminate()
                self._process.wait(timeout=5)
            except Exception:
                try:
                    self._process.kill()
                except Exception:
                    pass
        self._initialized = False
        with self._open_files_lock:
            self._open_files.clear()

    # ------------------------------------------------------------------
    # High-level LSP queries
    # ------------------------------------------------------------------

    def hover(self, file: str, line: int, column: int) -> Optional[dict]:
        """Get hover info at a location. Line/column are 1-based."""
        if not self.ensure_started():
            return None
        uri = self.ensure_file_open(file)
        result = self._send_request("textDocument/hover", {
            "textDocument": {"uri": uri},
            "position": {"line": line - 1, "character": column - 1},
        })
        if not result:
            return None

        contents = result.get("contents", {})
        if isinstance(contents, dict):
            return {"value": contents.get("value", ""), "kind": contents.get("kind", "")}
        elif isinstance(contents, str):
            return {"value": contents, "kind": "plaintext"}
        elif isinstance(contents, list):
            parts = []
            for c in contents:
                if isinstance(c, dict):
                    parts.append(c.get("value", ""))
                else:
                    parts.append(str(c))
            return {"value": "\n".join(parts), "kind": "plaintext"}
        return None

    def get_call_hierarchy(
        self, file: str, line: int, column: int, direction: str = "incoming"
    ) -> Optional[list[dict]]:
        """Get incoming or outgoing calls at a location."""
        if direction not in ("incoming", "outgoing"):
            raise ValueError(f"Invalid direction '{direction}'. Must be 'incoming' or 'outgoing'.")

        if not self.ensure_started():
            return None
        uri = self.ensure_file_open(file)

        # Step 1: prepareCallHierarchy
        items = self._send_request("textDocument/prepareCallHierarchy", {
            "textDocument": {"uri": uri},
            "position": {"line": line - 1, "character": column - 1},
        })
        if not items:
            return None

        # Step 2: get incoming or outgoing calls for each item
        results = []
        method = (
            "callHierarchy/incomingCalls" if direction == "incoming"
            else "callHierarchy/outgoingCalls"
        )
        for item in items:
            calls = self._send_request(method, {"item": item})
            if calls:
                for call in calls:
                    caller = call.get("from" if direction == "incoming" else "to", {})
                    results.append({
                        "name": caller.get("name", ""),
                        "kind": caller.get("kind", 0),
                        "file": self._uri_to_relative_path(caller.get("uri", "")),
                        "line": caller.get("range", {}).get("start", {}).get("line", 0) + 1,
                    })
        return results

    def get_type_hierarchy(
        self, file: str, line: int, column: int
    ) -> Optional[dict]:
        """Get type hierarchy (supertypes + subtypes) at a location."""
        if not self.ensure_started():
            return None
        uri = self.ensure_file_open(file)

        # prepareTypeHierarchy
        items = self._send_request("textDocument/prepareTypeHierarchy", {
            "textDocument": {"uri": uri},
            "position": {"line": line - 1, "character": column - 1},
        })
        if not items:
            return None

        result = {
            "name": items[0].get("name", "") if items else "",
            "supertypes": [],
            "subtypes": [],
        }

        for item in items:
            # Supertypes
            supers = self._send_request("typeHierarchy/supertypes", {"item": item})
            if supers:
                for s in supers:
                    result["supertypes"].append({
                        "name": s.get("name", ""),
                        "file": self._uri_to_relative_path(s.get("uri", "")),
                        "line": s.get("range", {}).get("start", {}).get("line", 0) + 1,
                    })

            # Subtypes
            subs = self._send_request("typeHierarchy/subtypes", {"item": item})
            if subs:
                for s in subs:
                    result["subtypes"].append({
                        "name": s.get("name", ""),
                        "file": self._uri_to_relative_path(s.get("uri", "")),
                        "line": s.get("range", {}).get("start", {}).get("line", 0) + 1,
                    })

        return result