# codex-lens LSP Integration - Complete Execution Plan > Version: 1.0 > Created: 2026-01-15 > Based on: Gemini Multi-Round Deep Analysis > Status: Ready for Execution --- ## Table of Contents 1. [Executive Summary](#1-executive-summary) 2. [Claude Code LSP Implementation Reference](#2-claude-code-lsp-implementation-reference) 3. [Architecture Overview](#3-architecture-overview) 4. [Phase 1: LSP Server Foundation](#4-phase-1-lsp-server-foundation) 5. [Phase 2: Find References](#5-phase-2-find-references) 6. [Phase 3: Hover Information](#6-phase-3-hover-information) 7. [Phase 4: MCP Bridge](#7-phase-4-mcp-bridge) 8. [Phase 5: Advanced Features](#8-phase-5-advanced-features) 9. [Testing Strategy](#9-testing-strategy) 10. [Deployment Guide](#10-deployment-guide) 11. [Risk Mitigation](#11-risk-mitigation) --- ## 1. Executive Summary ### 1.1 Project Goal 将 codex-lens 的代码索引和搜索能力通过 LSP (Language Server Protocol) 暴露,使其能够: - 为 IDE/编辑器提供代码导航功能 - 与 Claude Code 的 hook 系统集成 - 通过 MCP (Model Context Protocol) 为 LLM 提供结构化代码上下文 ### 1.2 Value Proposition | Capability | Before | After | |------------|--------|-------| | Code Navigation | CLI only | IDE integration via LSP | | Context for LLM | Manual copy-paste | Automated MCP injection | | Real-time Updates | Batch re-index | Incremental on save | | Cross-project Search | Per-directory | Unified global index | ### 1.3 Success Criteria - [ ] All 5 core LSP methods implemented and tested - [ ] Query latency < 100ms for 95th percentile - [ ] MCP context generation working with Claude Code hooks - [ ] Documentation and examples complete --- ## 2. Claude Code LSP Implementation Reference > 本章节记录 Claude Code 当前 LSP 实现方式,作为 codex-lens 集成的技术参考。 ### 2.1 Claude Code LSP 实现方式概览 Claude Code 实现 LSP 功能有 **三种方式**: | 方式 | 描述 | 适用场景 | |------|------|----------| | **内置 LSP 工具** | v2.0.74+ 原生支持 | 快速启用,基础功能 | | **MCP Server (cclsp)** | 第三方 MCP 桥接 | 高级功能,位置容错 | | **Plugin Marketplace** | 插件市场安装 | 多语言扩展支持 | ### 2.2 方式一:内置 LSP 工具 (v2.0.74+) Claude Code 从 v2.0.74 版本开始内置 LSP 支持。 #### 启用方式 ```bash # 设置环境变量启用 LSP export ENABLE_LSP_TOOL=1 claude # 永久启用 (添加到 shell 配置) echo 'export ENABLE_LSP_TOOL=1' >> ~/.bashrc ``` #### 内置 LSP 工具清单 | 工具名 | 功能 | 对应 LSP 方法 | 性能 | |--------|------|---------------|------| | `goToDefinition` | 跳转到符号定义 | `textDocument/definition` | ~50ms | | `findReferences` | 查找所有引用 | `textDocument/references` | ~100ms | | `documentSymbol` | 获取文件符号结构 | `textDocument/documentSymbol` | ~30ms | | `hover` | 显示类型签名和文档 | `textDocument/hover` | ~50ms | | `getDiagnostics` | 获取诊断信息 | `textDocument/diagnostic` | ~100ms | #### 性能对比 ``` 传统文本搜索: ~45,000ms (45秒) LSP 语义搜索: ~50ms 性能提升: 约 900 倍 ``` #### 当前限制 - 部分语言返回 "No LSP server available" - 需要额外安装语言服务器插件 - 不支持重命名等高级操作 ### 2.3 方式二:MCP Server 方式 (cclsp) [cclsp](https://github.com/ktnyt/cclsp) 是一个 MCP Server,将 LSP 能力暴露给 Claude Code。 #### 架构图 ``` ┌─────────────────────────────────────────────────────────────────┐ │ Claude Code │ │ (MCP Client) │ └───────────────────────────┬─────────────────────────────────────┘ │ │ MCP Protocol (JSON-RPC over stdio) │ ┌───────────────────────────▼─────────────────────────────────────┐ │ cclsp │ │ (MCP Server) │ │ ┌─────────────────────────────────────────────────────────┐ │ │ │ Position Tolerance Layer │ │ │ │ (自动尝试多个位置组合,解决 AI 行号不精确问题) │ │ │ └─────────────────────────────────────────────────────────┘ │ │ │ │ │ ┌──────────────────┼──────────────────┐ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ pylsp │ │ gopls │ │rust-analyzer│ │ │ │ (Python) │ │ (Go) │ │ (Rust) │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ │ └─────────────────────────────────────────────────────────────────┘ ``` #### 安装与配置 ```bash # 一次性运行 (无需安装) npx cclsp@latest setup # 用户级配置 npx cclsp@latest setup --user ``` #### 配置文件格式 **位置**: `.claude/cclsp.json` 或 `~/.config/claude/cclsp.json` ```json { "servers": [ { "extensions": ["py", "pyi"], "command": ["pylsp"], "rootDir": ".", "restartInterval": 5, "initializationOptions": {} }, { "extensions": ["ts", "tsx", "js", "jsx"], "command": ["typescript-language-server", "--stdio"], "rootDir": "." }, { "extensions": ["go"], "command": ["gopls"], "rootDir": "." }, { "extensions": ["rs"], "command": ["rust-analyzer"], "rootDir": "." } ] } ``` #### cclsp 暴露的 MCP 工具 | MCP 工具 | 功能 | 特性 | |----------|------|------| | `find_definition` | 按名称和类型查找定义 | 支持模糊匹配 | | `find_references` | 查找所有引用位置 | 跨文件搜索 | | `rename_symbol` | 重命名符号 | 创建 .bak 备份 | | `rename_symbol_strict` | 精确位置重命名 | 处理同名歧义 | | `get_diagnostics` | 获取诊断信息 | 错误/警告/提示 | | `restart_server` | 重启 LSP 服务器 | 解决内存泄漏 | #### 核心特性:位置容错 ```python # AI 生成的代码位置常有偏差 # cclsp 自动尝试多个位置组合 positions_to_try = [ (line, column), # 原始位置 (line - 1, column), # 上一行 (line + 1, column), # 下一行 (line, 0), # 行首 (line, len(line_content)) # 行尾 ] for pos in positions_to_try: result = lsp_server.definition(pos) if result: return result ``` #### 支持的语言服务器 | 语言 | 服务器 | 安装命令 | |------|--------|----------| | Python | pylsp | `pip install python-lsp-server` | | TypeScript | typescript-language-server | `npm i -g typescript-language-server` | | Go | gopls | `go install golang.org/x/tools/gopls@latest` | | Rust | rust-analyzer | `rustup component add rust-analyzer` | | C/C++ | clangd | `apt install clangd` | | Ruby | solargraph | `gem install solargraph` | | PHP | intelephense | `npm i -g intelephense` | | Java | jdtls | Eclipse JDT Language Server | ### 2.4 方式三:Plugin Marketplace 插件 Claude Code 官方插件市场提供语言支持扩展。 #### 添加插件市场 ```bash /plugin marketplace add boostvolt/claude-code-lsps ``` #### 安装语言支持 ```bash # Python (Pyright) /plugin install pyright@claude-code-lsps # TypeScript/JavaScript /plugin install vtsls@claude-code-lsps # Go /plugin install gopls@claude-code-lsps # Rust /plugin install rust-analyzer@claude-code-lsps # Java /plugin install jdtls@claude-code-lsps # C/C++ /plugin install clangd@claude-code-lsps # C# /plugin install omnisharp@claude-code-lsps # PHP /plugin install intelephense@claude-code-lsps # Kotlin /plugin install kotlin-language-server@claude-code-lsps # Ruby /plugin install solargraph@claude-code-lsps ``` #### 支持的 11 种语言 Python, TypeScript, Go, Rust, Java, C/C++, C#, PHP, Kotlin, Ruby, HTML/CSS ### 2.5 三种方式对比 | 特性 | 内置 LSP | cclsp (MCP) | Plugin Marketplace | |------|----------|-------------|-------------------| | 安装复杂度 | 低 (环境变量) | 中 (npx) | 低 (/plugin) | | 功能完整性 | 基础 5 个操作 | 完整 + 重命名 | 完整 | | 位置容错 | 无 | 有 | 无 | | 重命名支持 | 无 | 有 | 有 | | 自定义配置 | 无 | 完整 JSON | 有限 | | 多语言支持 | 需插件 | 任意 LSP | 11 种 | | 生产稳定性 | 高 | 中 | 高 | ### 2.6 codex-lens 集成策略 基于 Claude Code LSP 实现方式分析,推荐以下集成策略: #### 策略 A:作为 MCP Server (推荐) ``` ┌─────────────────────────────────────────────────────────────────┐ │ Claude Code │ └───────────────────────────┬─────────────────────────────────────┘ │ MCP Protocol ▼ ┌─────────────────────────────────────────────────────────────────┐ │ codex-lens MCP Server │ │ ┌─────────────────────────────────────────────────────────┐ │ │ │ MCP Tools │ │ │ │ • find_definition → GlobalSymbolIndex.search() │ │ │ │ • find_references → ChainSearchEngine.search_refs() │ │ │ │ • get_context → MCPProvider.build_context() │ │ │ │ • hybrid_search → HybridSearchEngine.search() │ │ │ └─────────────────────────────────────────────────────────┘ │ │ │ │ │ ┌─────────────────────────▼───────────────────────────────┐ │ │ │ codex-lens Core │ │ │ │ GlobalSymbolIndex │ SQLiteStore │ WatcherManager │ │ │ └─────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────┘ ``` **优势**: - 直接复用 codex-lens 索引 - 无需启动额外 LSP 进程 - 支持 MCP 上下文注入 **实现文件**: `src/codexlens/mcp/server.py` ```python """codex-lens MCP Server for Claude Code integration.""" import json import sys from typing import Any, Dict from codexlens.mcp.provider import MCPProvider from codexlens.search.chain_search import ChainSearchEngine from codexlens.storage.global_index import GlobalSymbolIndex class CodexLensMCPServer: """MCP Server exposing codex-lens capabilities.""" def __init__(self, workspace_path: str): self.global_index = GlobalSymbolIndex(workspace_path) self.search_engine = ChainSearchEngine(...) self.mcp_provider = MCPProvider(...) def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]: """Handle MCP tool call.""" method = request.get("method") params = request.get("params", {}) handlers = { "find_definition": self._find_definition, "find_references": self._find_references, "get_context": self._get_context, "hybrid_search": self._hybrid_search, } handler = handlers.get(method) if handler: return handler(params) return {"error": f"Unknown method: {method}"} def _find_definition(self, params: Dict) -> Dict: """Find symbol definition.""" symbol_name = params.get("symbol") symbols = self.global_index.search(symbol_name, exact=True, limit=1) if symbols: s = symbols[0] return { "file": s.file_path, "line": s.range[0], "column": 0, "kind": s.kind, } return {"error": "Symbol not found"} def _find_references(self, params: Dict) -> Dict: """Find all references.""" symbol_name = params.get("symbol") refs = self.search_engine.search_references(symbol_name) return { "references": [ {"file": r.file_path, "line": r.line, "context": r.context} for r in refs ] } def _get_context(self, params: Dict) -> Dict: """Get MCP context for LLM.""" symbol_name = params.get("symbol") context = self.mcp_provider.build_context(symbol_name) return context.to_dict() if context else {"error": "Context not found"} def _hybrid_search(self, params: Dict) -> Dict: """Execute hybrid search.""" query = params.get("query") # ... implementation ``` #### 策略 B:作为独立 LSP Server 通过 cclsp 配置接入 codex-lens LSP Server。 **cclsp 配置** (`.claude/cclsp.json`): ```json { "servers": [ { "extensions": ["py", "ts", "go", "rs", "java"], "command": ["codexlens-lsp", "--stdio"], "rootDir": ".", "restartInterval": 0 } ] } ``` **优势**: - 兼容标准 LSP 协议 - 可被任意 LSP 客户端使用 - cclsp 提供位置容错 #### 策略 C:混合模式 (最佳实践) ``` ┌───────────────────────────────────────────────────────────────────┐ │ Claude Code │ │ ┌──────────────────┐ ┌──────────────────────────┐ │ │ │ 内置 LSP 工具 │ │ MCP Client │ │ │ │ (基础导航) │ │ (上下文注入) │ │ │ └────────┬─────────┘ └────────────┬─────────────┘ │ └───────────┼───────────────────────────────────┼──────────────────┘ │ │ │ LSP Protocol │ MCP Protocol │ │ ┌───────────▼───────────────────────────────────▼──────────────────┐ │ codex-lens Unified Server │ │ ┌─────────────────────────┐ ┌─────────────────────────────┐ │ │ │ LSP Handlers │ │ MCP Handlers │ │ │ │ • definition │ │ • get_context │ │ │ │ • references │ │ • enrich_prompt │ │ │ │ • hover │ │ • hybrid_search │ │ │ │ • completion │ │ • semantic_query │ │ │ └────────────┬────────────┘ └──────────────┬──────────────┘ │ │ │ │ │ │ └───────────────┬───────────────┘ │ │ ▼ │ │ ┌─────────────────────────────────────────────────────────────┐ │ │ │ codex-lens Core │ │ │ │ GlobalSymbolIndex │ HybridSearch │ VectorStore │ Watcher │ │ │ └─────────────────────────────────────────────────────────────┘ │ └──────────────────────────────────────────────────────────────────┘ ``` **优势**: - LSP 提供标准代码导航 - MCP 提供 LLM 上下文增强 - 统一索引,避免重复计算 ### 2.7 参考资源 | 资源 | 链接 | |------|------| | Claude Code LSP 设置指南 | https://www.aifreeapi.com/en/posts/claude-code-lsp | | cclsp GitHub | https://github.com/ktnyt/cclsp | | Claude Code Plugins | https://code.claude.com/docs/en/plugins-reference | | claude-code-lsps 市场 | https://github.com/Piebald-AI/claude-code-lsps | | LSP 规范 | https://microsoft.github.io/language-server-protocol/ | | MCP 规范 | https://modelcontextprotocol.io/ | --- ## 3. Architecture Overview ### 3.1 Target Architecture ``` ┌─────────────────────────────────────────────────────────────────────────────┐ │ Client Layer │ ├─────────────────────────────────────────────────────────────────────────────┤ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │ │ │ VS Code │ │ Neovim │ │ Sublime │ │ Claude Code │ │ │ │ (LSP Client)│ │ (LSP Client)│ │ (LSP Client)│ │ (Hook + MCP Client) │ │ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ └──────────┬──────────┘ │ │ │ │ │ │ │ └─────────┼────────────────┼────────────────┼─────────────────────┼───────────┘ │ │ │ │ └────────────────┴────────────────┴──────────┬──────────┘ │ (JSON-RPC / stdio) │ ┌──────────────────────────────────────────────────────┴──────────────────────┐ │ codex-lens LSP Server │ ├─────────────────────────────────────────────────────────────────────────────┤ │ │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ LSP Layer (NEW) │ │ │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ │ │ Handlers │ │ Providers │ │ Protocol │ │ │ │ │ │ definition │ │ hover │ │ messages │ │ │ │ │ │ references │ │ completion │ │ lifecycle │ │ │ │ │ │ symbols │ │ │ │ │ │ │ │ │ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │ │ │ └─────────┼─────────────────┼─────────────────┼───────────────────────┘ │ │ │ │ │ │ │ ┌─────────┴─────────────────┴─────────────────┴───────────────────────┐ │ │ │ MCP Layer (NEW) │ │ │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ │ │ Schema │ │ Provider │ │ Hooks │ │ │ │ │ │ MCPContext │ │ buildContext │ │ pre-tool │ │ │ │ │ │ SymbolInfo │ │ enrichPrompt │ │ post-tool │ │ │ │ │ └──────────────┘ └──────────────┘ └──────────────┘ │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ │ │ │ │ ┌─────────────────────────────────┴───────────────────────────────────┐ │ │ │ Existing codex-lens Core │ │ │ │ │ │ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌────────────┐ │ │ │ │ │ Search │ │ Storage │ │ Watcher │ │ Parser │ │ │ │ │ │ ChainSearch │ │ GlobalIndex │ │ Manager │ │ TreeSitter │ │ │ │ │ │ HybridSearch│ │ SQLiteStore │ │ Incremental │ │ Symbols │ │ │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ └────────────┘ │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────────────────────────┘ ``` ### 3.2 Data Flow ``` LSP Request Flow ================ [Client] ─── textDocument/definition ───> [LSP Server] │ v ┌─────────────────┐ │ Parse Request │ │ Extract symbol │ └────────┬────────┘ │ v ┌─────────────────┐ │ GlobalSymbolIdx │ │ .search() │ └────────┬────────┘ │ v ┌─────────────────┐ │ Format Result │ │ as Location │ └────────┬────────┘ │ [Client] <─── Location Response ────────────────┘ MCP Context Flow ================ [Claude Code] ─── pre-tool hook ───> [MCP Provider] │ ┌─────────────────────┴─────────────────────┐ v v v ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ Definition │ │ References │ │ Related │ │ Lookup │ │ Lookup │ │ Symbols │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │ │ └─────────────────────┴─────────────────────┘ │ v ┌───────────────┐ │ MCPContext │ │ Object │ └───────┬───────┘ │ [Claude Code] <─── JSON Context ──────────┘ │ v ┌───────────────────────┐ │ Inject into LLM Prompt│ └───────────────────────┘ ``` ### 3.3 Module Dependencies ``` ┌─────────────────────┐ │ lsp/server.py │ │ (Entry Point) │ └──────────┬──────────┘ │ ┌───────────────────┼───────────────────┐ │ │ │ v v v ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │lsp/handlers │ │lsp/providers│ │ mcp/provider│ │ .py │ │ .py │ │ .py │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │ │ └───────────────────┼───────────────────┘ │ v ┌─────────────────────┐ │ search/chain_search │ │ .py │ └──────────┬──────────┘ │ ┌───────────────────┼───────────────────┐ │ │ │ v v v ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │storage/ │ │storage/ │ │watcher/ │ │global_index │ │sqlite_store │ │manager.py │ └─────────────┘ └─────────────┘ └─────────────┘ ``` --- ## 4. Phase 1: LSP Server Foundation ### 4.1 Overview | Attribute | Value | |-----------|-------| | Priority | HIGH | | Complexity | Medium | | Dependencies | pygls library | | Deliverables | Working LSP server with 3 core handlers | ### 4.2 Task Breakdown #### Task 1.1: Project Setup **File**: `pyproject.toml` (MODIFY) ```toml [project.optional-dependencies] lsp = [ "pygls>=1.3.0", ] [project.scripts] codexlens-lsp = "codexlens.lsp:main" ``` **Acceptance Criteria**: - [ ] `pip install -e ".[lsp]"` succeeds - [ ] `codexlens-lsp --help` shows usage --- #### Task 1.2: LSP Server Core **File**: `src/codexlens/lsp/__init__.py` (NEW) ```python """codex-lens Language Server Protocol implementation.""" from codexlens.lsp.server import CodexLensLanguageServer, main __all__ = ["CodexLensLanguageServer", "main"] ``` **File**: `src/codexlens/lsp/server.py` (NEW) ```python """Main LSP server implementation using pygls.""" import logging from pathlib import Path from typing import Optional from lsprotocol import types as lsp from pygls.server import LanguageServer from codexlens.search.chain_search import ChainSearchEngine from codexlens.storage.global_index import GlobalSymbolIndex from codexlens.storage.registry import RegistryStore from codexlens.storage.path_mapper import PathMapper from codexlens.watcher.manager import WatcherManager logger = logging.getLogger(__name__) class CodexLensLanguageServer(LanguageServer): """Language Server powered by codex-lens indexing.""" def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.workspace_path: Optional[Path] = None self.registry: Optional[RegistryStore] = None self.search_engine: Optional[ChainSearchEngine] = None self.global_index: Optional[GlobalSymbolIndex] = None self.watcher: Optional[WatcherManager] = None def initialize_codexlens(self, workspace_path: Path) -> None: """Initialize codex-lens components for the workspace.""" self.workspace_path = workspace_path # Initialize registry and search engine self.registry = RegistryStore() self.registry.initialize() mapper = PathMapper() self.search_engine = ChainSearchEngine(self.registry, mapper) # Initialize global symbol index self.global_index = GlobalSymbolIndex(workspace_path) # Start file watcher for incremental updates self.watcher = WatcherManager( root_path=workspace_path, on_indexed=self._on_file_indexed ) self.watcher.start() logger.info(f"Initialized codex-lens for workspace: {workspace_path}") def _on_file_indexed(self, file_path: Path) -> None: """Callback when a file is indexed.""" logger.debug(f"File indexed: {file_path}") def shutdown_codexlens(self) -> None: """Cleanup codex-lens components.""" if self.watcher: self.watcher.stop() self.watcher = None logger.info("codex-lens shutdown complete") # Create server instance server = CodexLensLanguageServer( name="codex-lens", version="0.1.0" ) @server.feature(lsp.INITIALIZE) def on_initialize(params: lsp.InitializeParams) -> lsp.InitializeResult: """Handle LSP initialize request.""" if params.root_uri: workspace_path = Path(params.root_uri.replace("file://", "")) server.initialize_codexlens(workspace_path) return lsp.InitializeResult( capabilities=lsp.ServerCapabilities( text_document_sync=lsp.TextDocumentSyncOptions( open_close=True, change=lsp.TextDocumentSyncKind.Incremental, save=lsp.SaveOptions(include_text=False), ), definition_provider=True, references_provider=True, completion_provider=lsp.CompletionOptions( trigger_characters=[".", "_"], ), hover_provider=True, workspace_symbol_provider=True, ), server_info=lsp.ServerInfo( name="codex-lens", version="0.1.0", ), ) @server.feature(lsp.SHUTDOWN) def on_shutdown(params: None) -> None: """Handle LSP shutdown request.""" server.shutdown_codexlens() def main(): """Entry point for the LSP server.""" import argparse parser = argparse.ArgumentParser(description="codex-lens Language Server") parser.add_argument("--stdio", action="store_true", help="Use stdio transport") parser.add_argument("--tcp", action="store_true", help="Use TCP transport") parser.add_argument("--host", default="127.0.0.1", help="TCP host") parser.add_argument("--port", type=int, default=2087, help="TCP port") args = parser.parse_args() if args.tcp: server.start_tcp(args.host, args.port) else: server.start_io() if __name__ == "__main__": main() ``` **Acceptance Criteria**: - [ ] Server starts without errors - [ ] Handles initialize/shutdown lifecycle - [ ] WatcherManager starts on workspace open --- #### Task 1.3: Definition Handler **File**: `src/codexlens/lsp/handlers.py` (NEW) ```python """LSP request handlers.""" import logging from pathlib import Path from typing import List, Optional, Union from lsprotocol import types as lsp from codexlens.lsp.server import server from codexlens.entities import Symbol logger = logging.getLogger(__name__) def symbol_to_location(symbol: Symbol) -> lsp.Location: """Convert codex-lens Symbol to LSP Location.""" return lsp.Location( uri=f"file://{symbol.file_path}", range=lsp.Range( start=lsp.Position( line=symbol.range[0] - 1, # LSP is 0-indexed character=0, ), end=lsp.Position( line=symbol.range[1] - 1, character=0, ), ), ) @server.feature(lsp.TEXT_DOCUMENT_DEFINITION) def on_definition( params: lsp.DefinitionParams, ) -> Optional[Union[lsp.Location, List[lsp.Location]]]: """Handle textDocument/definition request.""" if not server.global_index: return None # Get the word at cursor position document = server.workspace.get_text_document(params.text_document.uri) word = _get_word_at_position(document, params.position) if not word: return None logger.debug(f"Definition lookup for: {word}") # Search in global symbol index symbols = server.global_index.search(word, exact=True, limit=10) if not symbols: return None if len(symbols) == 1: return symbol_to_location(symbols[0]) return [symbol_to_location(s) for s in symbols] def _get_word_at_position(document, position: lsp.Position) -> Optional[str]: """Extract the word at the given position.""" try: lines = document.source.split("\n") if position.line >= len(lines): return None line = lines[position.line] # Find word boundaries start = position.character end = position.character # Expand left while start > 0 and _is_identifier_char(line[start - 1]): start -= 1 # Expand right while end < len(line) and _is_identifier_char(line[end]): end += 1 word = line[start:end] return word if word else None except Exception as e: logger.error(f"Error extracting word: {e}") return None def _is_identifier_char(char: str) -> bool: """Check if character is valid in an identifier.""" return char.isalnum() or char == "_" ``` **Acceptance Criteria**: - [ ] Returns Location for known symbols - [ ] Returns None for unknown symbols - [ ] Handles multiple definitions (overloads) --- #### Task 1.4: Completion Handler **File**: `src/codexlens/lsp/handlers.py` (APPEND) ```python @server.feature(lsp.TEXT_DOCUMENT_COMPLETION) def on_completion( params: lsp.CompletionParams, ) -> Optional[lsp.CompletionList]: """Handle textDocument/completion request.""" if not server.global_index: return None # Get partial word at cursor document = server.workspace.get_text_document(params.text_document.uri) prefix = _get_prefix_at_position(document, params.position) if not prefix or len(prefix) < 2: return None logger.debug(f"Completion lookup for prefix: {prefix}") # Search with prefix mode symbols = server.global_index.search(prefix, prefix_mode=True, limit=50) if not symbols: return None items = [] for symbol in symbols: kind = _symbol_kind_to_completion_kind(symbol.kind) items.append( lsp.CompletionItem( label=symbol.name, kind=kind, detail=f"{symbol.kind} in {Path(symbol.file_path).name}", documentation=lsp.MarkupContent( kind=lsp.MarkupKind.Markdown, value=f"Defined at line {symbol.range[0]}", ), ) ) return lsp.CompletionList(is_incomplete=len(items) >= 50, items=items) def _get_prefix_at_position(document, position: lsp.Position) -> Optional[str]: """Extract the incomplete word prefix at position.""" try: lines = document.source.split("\n") if position.line >= len(lines): return None line = lines[position.line] # Find prefix start start = position.character while start > 0 and _is_identifier_char(line[start - 1]): start -= 1 return line[start:position.character] if start < position.character else None except Exception: return None def _symbol_kind_to_completion_kind(kind: str) -> lsp.CompletionItemKind: """Map symbol kind to LSP completion kind.""" mapping = { "function": lsp.CompletionItemKind.Function, "method": lsp.CompletionItemKind.Method, "class": lsp.CompletionItemKind.Class, "variable": lsp.CompletionItemKind.Variable, "constant": lsp.CompletionItemKind.Constant, "module": lsp.CompletionItemKind.Module, "property": lsp.CompletionItemKind.Property, "interface": lsp.CompletionItemKind.Interface, "enum": lsp.CompletionItemKind.Enum, } return mapping.get(kind.lower(), lsp.CompletionItemKind.Text) ``` **Acceptance Criteria**: - [ ] Returns completion items for valid prefixes - [ ] Respects minimum prefix length (2 chars) - [ ] Maps symbol kinds correctly --- #### Task 1.5: Workspace Symbol Handler **File**: `src/codexlens/lsp/handlers.py` (APPEND) ```python @server.feature(lsp.WORKSPACE_SYMBOL) def on_workspace_symbol( params: lsp.WorkspaceSymbolParams, ) -> Optional[List[lsp.SymbolInformation]]: """Handle workspace/symbol request.""" if not server.search_engine or not server.workspace_path: return None query = params.query if not query or len(query) < 2: return None logger.debug(f"Workspace symbol search: {query}") # Use chain search engine's symbol search result = server.search_engine.search_symbols( query=query, source_path=server.workspace_path, limit=100, ) if not result: return None items = [] for symbol in result: kind = _symbol_kind_to_symbol_kind(symbol.kind) items.append( lsp.SymbolInformation( name=symbol.name, kind=kind, location=symbol_to_location(symbol), container_name=Path(symbol.file_path).parent.name, ) ) return items def _symbol_kind_to_symbol_kind(kind: str) -> lsp.SymbolKind: """Map symbol kind string to LSP SymbolKind.""" mapping = { "function": lsp.SymbolKind.Function, "method": lsp.SymbolKind.Method, "class": lsp.SymbolKind.Class, "variable": lsp.SymbolKind.Variable, "constant": lsp.SymbolKind.Constant, "module": lsp.SymbolKind.Module, "property": lsp.SymbolKind.Property, "interface": lsp.SymbolKind.Interface, "enum": lsp.SymbolKind.Enum, "struct": lsp.SymbolKind.Struct, "namespace": lsp.SymbolKind.Namespace, } return mapping.get(kind.lower(), lsp.SymbolKind.Variable) ``` **Acceptance Criteria**: - [ ] Returns symbols matching query - [ ] Respects result limit - [ ] Includes container information --- #### Task 1.6: File Watcher Integration **File**: `src/codexlens/lsp/handlers.py` (APPEND) ```python @server.feature(lsp.TEXT_DOCUMENT_DID_SAVE) def on_did_save(params: lsp.DidSaveTextDocumentParams) -> None: """Handle textDocument/didSave notification.""" if not server.watcher: return file_path = Path(params.text_document.uri.replace("file://", "")) logger.debug(f"File saved: {file_path}") # Trigger incremental indexing server.watcher.trigger_index(file_path) @server.feature(lsp.TEXT_DOCUMENT_DID_OPEN) def on_did_open(params: lsp.DidOpenTextDocumentParams) -> None: """Handle textDocument/didOpen notification.""" logger.debug(f"File opened: {params.text_document.uri}") @server.feature(lsp.TEXT_DOCUMENT_DID_CLOSE) def on_did_close(params: lsp.DidCloseTextDocumentParams) -> None: """Handle textDocument/didClose notification.""" logger.debug(f"File closed: {params.text_document.uri}") ``` **Acceptance Criteria**: - [ ] didSave triggers incremental index - [ ] No blocking on save - [ ] Proper logging --- ### 4.3 Phase 1 Test Plan **File**: `tests/lsp/test_server.py` (NEW) ```python """Tests for LSP server.""" import pytest from pathlib import Path from unittest.mock import Mock, patch from lsprotocol import types as lsp from codexlens.lsp.server import CodexLensLanguageServer, on_initialize class TestServerInitialization: """Test server lifecycle.""" def test_initialize_creates_components(self, tmp_path): """Server creates all components on initialize.""" server = CodexLensLanguageServer("test", "0.1.0") params = lsp.InitializeParams( root_uri=f"file://{tmp_path}", capabilities=lsp.ClientCapabilities(), ) result = on_initialize(params) assert result.capabilities.definition_provider assert result.capabilities.completion_provider assert result.capabilities.workspace_symbol_provider class TestDefinitionHandler: """Test textDocument/definition handler.""" def test_definition_returns_location(self): """Definition returns valid Location.""" # Setup mock global index mock_symbol = Mock() mock_symbol.file_path = "/test/file.py" mock_symbol.range = (10, 15) with patch.object(server, 'global_index') as mock_index: mock_index.search.return_value = [mock_symbol] # Call handler result = on_definition(Mock( text_document=Mock(uri="file:///test/file.py"), position=lsp.Position(line=5, character=10), )) assert isinstance(result, lsp.Location) assert result.uri == "file:///test/file.py" class TestCompletionHandler: """Test textDocument/completion handler.""" def test_completion_returns_items(self): """Completion returns CompletionList.""" # Test implementation pass ``` **Acceptance Criteria**: - [ ] All unit tests pass - [ ] Coverage > 80% for LSP module - [ ] Integration test with real workspace --- ## 5. Phase 2: Find References ### 5.1 Overview | Attribute | Value | |-----------|-------| | Priority | MEDIUM | | Complexity | High | | Dependencies | Phase 1 complete | | Deliverables | `search_references()` method + LSP handler | ### 5.2 Task Breakdown #### Task 2.1: Add `search_references` to ChainSearchEngine **File**: `src/codexlens/search/chain_search.py` (MODIFY) ```python # Add to ChainSearchEngine class from dataclasses import dataclass from typing import List from concurrent.futures import ThreadPoolExecutor, as_completed @dataclass class ReferenceResult: """Result from reference search.""" file_path: str line: int column: int context: str # Surrounding code snippet relationship_type: str # "call", "import", "inheritance", etc. def search_references( self, symbol_name: str, source_path: Optional[Path] = None, depth: int = -1, limit: int = 100, ) -> List[ReferenceResult]: """Find all references to a symbol across the project. Args: symbol_name: Fully qualified or simple name of the symbol source_path: Starting path for search (default: workspace root) depth: Search depth (-1 = unlimited) limit: Maximum results to return Returns: List of ReferenceResult objects sorted by file path and line """ source = source_path or self._workspace_path # Collect all index paths index_paths = self._collect_index_paths(source, depth) if not index_paths: logger.warning(f"No indexes found for reference search: {source}") return [] # Parallel query across all indexes all_results: List[ReferenceResult] = [] with ThreadPoolExecutor(max_workers=self._options.max_workers) as executor: futures = { executor.submit( self._search_references_single, idx_path, symbol_name, ): idx_path for idx_path in index_paths } for future in as_completed(futures): try: results = future.result(timeout=10) all_results.extend(results) except Exception as e: logger.error(f"Reference search failed: {e}") # Sort and limit all_results.sort(key=lambda r: (r.file_path, r.line)) return all_results[:limit] def _search_references_single( self, index_path: Path, symbol_name: str, ) -> List[ReferenceResult]: """Search for references in a single index.""" results = [] try: store = DirIndexStore(index_path.parent) # Query code_relationships table query = """ SELECT cr.source_file, cr.source_line, cr.source_column, cr.relationship_type, f.content FROM code_relationships cr JOIN files f ON f.full_path = cr.source_file WHERE cr.target_qualified_name LIKE ? OR cr.target_name = ? ORDER BY cr.source_file, cr.source_line """ rows = store.execute_query( query, (f"%{symbol_name}", symbol_name), ) for row in rows: # Extract context (3 lines around reference) content_lines = row["content"].split("\n") line_idx = row["source_line"] - 1 start = max(0, line_idx - 1) end = min(len(content_lines), line_idx + 2) context = "\n".join(content_lines[start:end]) results.append(ReferenceResult( file_path=row["source_file"], line=row["source_line"], column=row["source_column"] or 0, context=context, relationship_type=row["relationship_type"], )) except Exception as e: logger.error(f"Failed to search references in {index_path}: {e}") return results ``` **Acceptance Criteria**: - [ ] Searches all index files in parallel - [ ] Returns properly formatted ReferenceResult - [ ] Handles missing indexes gracefully --- #### Task 2.2: LSP References Handler **File**: `src/codexlens/lsp/handlers.py` (APPEND) ```python @server.feature(lsp.TEXT_DOCUMENT_REFERENCES) def on_references( params: lsp.ReferenceParams, ) -> Optional[List[lsp.Location]]: """Handle textDocument/references request.""" if not server.search_engine or not server.workspace_path: return None # Get the word at cursor document = server.workspace.get_text_document(params.text_document.uri) word = _get_word_at_position(document, params.position) if not word: return None logger.debug(f"References lookup for: {word}") # Search for references references = server.search_engine.search_references( symbol_name=word, source_path=server.workspace_path, limit=200, ) if not references: return None # Convert to LSP Locations locations = [] for ref in references: locations.append( lsp.Location( uri=f"file://{ref.file_path}", range=lsp.Range( start=lsp.Position(line=ref.line - 1, character=ref.column), end=lsp.Position(line=ref.line - 1, character=ref.column + len(word)), ), ) ) return locations ``` **Acceptance Criteria**: - [ ] Returns all references across project - [ ] Includes definition if `params.context.include_declaration` - [ ] Performance < 200ms for typical project --- ### 5.3 Phase 2 Test Plan ```python class TestReferencesSearch: """Test reference search functionality.""" def test_finds_function_calls(self, indexed_project): """Finds all calls to a function.""" results = search_engine.search_references("my_function") assert len(results) > 0 assert all(r.relationship_type == "call" for r in results) def test_finds_imports(self, indexed_project): """Finds all imports of a module.""" results = search_engine.search_references("my_module") assert any(r.relationship_type == "import" for r in results) def test_parallel_search_performance(self, large_project): """Parallel search completes within time limit.""" import time start = time.time() results = search_engine.search_references("common_symbol") elapsed = time.time() - start assert elapsed < 0.2 # 200ms ``` --- ## 6. Phase 3: Hover Information ### 6.1 Overview | Attribute | Value | |-----------|-------| | Priority | MEDIUM | | Complexity | Low | | Dependencies | Phase 1 complete | | Deliverables | Hover provider + LSP handler | ### 6.2 Task Breakdown #### Task 3.1: Hover Provider **File**: `src/codexlens/lsp/providers.py` (NEW) ```python """LSP feature providers.""" import logging from dataclasses import dataclass from pathlib import Path from typing import Optional from codexlens.entities import Symbol from codexlens.storage.sqlite_store import SQLiteStore logger = logging.getLogger(__name__) @dataclass class HoverInfo: """Hover information for a symbol.""" name: str kind: str signature: str documentation: Optional[str] file_path: str line_range: tuple class HoverProvider: """Provides hover information for symbols.""" def __init__(self, global_index, registry): self.global_index = global_index self.registry = registry def get_hover_info(self, symbol_name: str) -> Optional[HoverInfo]: """Get hover information for a symbol. Args: symbol_name: Name of the symbol to look up Returns: HoverInfo or None if symbol not found """ # Look up symbol in global index symbols = self.global_index.search(symbol_name, exact=True, limit=1) if not symbols: return None symbol = symbols[0] # Extract signature from source signature = self._extract_signature(symbol) return HoverInfo( name=symbol.name, kind=symbol.kind, signature=signature, documentation=symbol.docstring, file_path=symbol.file_path, line_range=symbol.range, ) def _extract_signature(self, symbol: Symbol) -> str: """Extract function/class signature from source.""" try: # Find the index for this file index_path = self.registry.find_index_path( Path(symbol.file_path).parent ) if not index_path: return f"{symbol.kind} {symbol.name}" store = SQLiteStore(index_path.parent) # Get file content rows = store.execute_query( "SELECT content FROM files WHERE full_path = ?", (symbol.file_path,), ) if not rows: return f"{symbol.kind} {symbol.name}" content = rows[0]["content"] lines = content.split("\n") # Extract signature lines start_line = symbol.range[0] - 1 signature_lines = [] # Get first line (def/class declaration) if start_line < len(lines): first_line = lines[start_line] signature_lines.append(first_line) # Continue if line ends with backslash or doesn't have closing paren i = start_line + 1 while i < len(lines) and i < start_line + 5: if "):" in signature_lines[-1] or ":" in signature_lines[-1]: break signature_lines.append(lines[i]) i += 1 return "\n".join(signature_lines) except Exception as e: logger.error(f"Failed to extract signature: {e}") return f"{symbol.kind} {symbol.name}" def format_hover_markdown(self, info: HoverInfo) -> str: """Format hover info as Markdown.""" parts = [] # Code block with signature parts.append(f"```python\n{info.signature}\n```") # Documentation if available if info.documentation: parts.append(f"\n---\n\n{info.documentation}") # Location info parts.append( f"\n---\n\n*{info.kind}* defined in " f"`{Path(info.file_path).name}` " f"(line {info.line_range[0]})" ) return "\n".join(parts) ``` --- #### Task 3.2: LSP Hover Handler **File**: `src/codexlens/lsp/handlers.py` (APPEND) ```python from codexlens.lsp.providers import HoverProvider @server.feature(lsp.TEXT_DOCUMENT_HOVER) def on_hover(params: lsp.HoverParams) -> Optional[lsp.Hover]: """Handle textDocument/hover request.""" if not server.global_index or not server.registry: return None # Get word at cursor document = server.workspace.get_text_document(params.text_document.uri) word = _get_word_at_position(document, params.position) if not word: return None logger.debug(f"Hover lookup for: {word}") # Get hover info provider = HoverProvider(server.global_index, server.registry) info = provider.get_hover_info(word) if not info: return None # Format as markdown content = provider.format_hover_markdown(info) return lsp.Hover( contents=lsp.MarkupContent( kind=lsp.MarkupKind.Markdown, value=content, ), ) ``` **Acceptance Criteria**: - [ ] Shows function signature - [ ] Shows documentation if available - [ ] Shows file location --- ## 7. Phase 4: MCP Bridge ### 7.1 Overview | Attribute | Value | |-----------|-------| | Priority | HIGH VALUE | | Complexity | Medium | | Dependencies | Phase 1-2 complete | | Deliverables | MCP schema + provider + hook interfaces | ### 7.2 Task Breakdown #### Task 4.1: MCP Schema Definition **File**: `src/codexlens/mcp/__init__.py` (NEW) ```python """Model Context Protocol implementation for Claude Code integration.""" from codexlens.mcp.schema import ( MCPContext, SymbolInfo, ReferenceInfo, RelatedSymbol, ) from codexlens.mcp.provider import MCPProvider __all__ = [ "MCPContext", "SymbolInfo", "ReferenceInfo", "RelatedSymbol", "MCPProvider", ] ``` **File**: `src/codexlens/mcp/schema.py` (NEW) ```python """MCP data models.""" from dataclasses import dataclass, field, asdict from typing import List, Optional import json @dataclass class SymbolInfo: """Information about a code symbol.""" name: str kind: str file_path: str line_start: int line_end: int signature: Optional[str] = None documentation: Optional[str] = None def to_dict(self) -> dict: return asdict(self) @dataclass class ReferenceInfo: """Information about a symbol reference.""" file_path: str line: int column: int context: str relationship_type: str def to_dict(self) -> dict: return asdict(self) @dataclass class RelatedSymbol: """Related symbol (import, call target, etc.).""" name: str kind: str relationship: str # "imports", "calls", "inherits", "uses" file_path: Optional[str] = None def to_dict(self) -> dict: return asdict(self) @dataclass class MCPContext: """Model Context Protocol context object. This is the structured context that gets injected into LLM prompts to provide code understanding. """ version: str = "1.0" context_type: str = "code_context" symbol: Optional[SymbolInfo] = None definition: Optional[str] = None references: List[ReferenceInfo] = field(default_factory=list) related_symbols: List[RelatedSymbol] = field(default_factory=list) metadata: dict = field(default_factory=dict) def to_dict(self) -> dict: """Convert to dictionary for JSON serialization.""" result = { "version": self.version, "context_type": self.context_type, "metadata": self.metadata, } if self.symbol: result["symbol"] = self.symbol.to_dict() if self.definition: result["definition"] = self.definition if self.references: result["references"] = [r.to_dict() for r in self.references] if self.related_symbols: result["related_symbols"] = [s.to_dict() for s in self.related_symbols] return result def to_json(self, indent: int = 2) -> str: """Serialize to JSON string.""" return json.dumps(self.to_dict(), indent=indent) def to_prompt_injection(self) -> str: """Format for injection into LLM prompt.""" parts = [""] if self.symbol: parts.append(f"## Symbol: {self.symbol.name}") parts.append(f"Type: {self.symbol.kind}") parts.append(f"Location: {self.symbol.file_path}:{self.symbol.line_start}") if self.definition: parts.append("\n## Definition") parts.append(f"```\n{self.definition}\n```") if self.references: parts.append(f"\n## References ({len(self.references)} found)") for i, ref in enumerate(self.references[:5]): # Limit to 5 parts.append(f"- {ref.file_path}:{ref.line} ({ref.relationship_type})") parts.append(f" ```\n {ref.context}\n ```") if self.related_symbols: parts.append("\n## Related Symbols") for sym in self.related_symbols[:10]: # Limit to 10 parts.append(f"- {sym.name} ({sym.relationship})") parts.append("") return "\n".join(parts) ``` --- #### Task 4.2: MCP Provider **File**: `src/codexlens/mcp/provider.py` (NEW) ```python """MCP context provider.""" import logging from pathlib import Path from typing import Optional, List from codexlens.mcp.schema import ( MCPContext, SymbolInfo, ReferenceInfo, RelatedSymbol, ) from codexlens.search.chain_search import ChainSearchEngine from codexlens.storage.global_index import GlobalSymbolIndex from codexlens.storage.sqlite_store import SQLiteStore from codexlens.storage.registry import RegistryStore logger = logging.getLogger(__name__) class MCPProvider: """Builds MCP context objects from codex-lens data.""" def __init__( self, global_index: GlobalSymbolIndex, search_engine: ChainSearchEngine, registry: RegistryStore, ): self.global_index = global_index self.search_engine = search_engine self.registry = registry def build_context( self, symbol_name: str, context_type: str = "symbol_explanation", include_references: bool = True, include_related: bool = True, max_references: int = 10, ) -> Optional[MCPContext]: """Build comprehensive context for a symbol. Args: symbol_name: Name of the symbol to contextualize context_type: Type of context being requested include_references: Whether to include reference locations include_related: Whether to include related symbols max_references: Maximum number of references to include Returns: MCPContext object or None if symbol not found """ # Look up symbol symbols = self.global_index.search(symbol_name, exact=True, limit=1) if not symbols: logger.warning(f"Symbol not found for MCP context: {symbol_name}") return None symbol = symbols[0] # Build SymbolInfo symbol_info = SymbolInfo( name=symbol.name, kind=symbol.kind, file_path=symbol.file_path, line_start=symbol.range[0], line_end=symbol.range[1], signature=getattr(symbol, 'signature', None), documentation=getattr(symbol, 'docstring', None), ) # Extract definition source code definition = self._extract_definition(symbol) # Get references references = [] if include_references: refs = self.search_engine.search_references( symbol_name, limit=max_references, ) references = [ ReferenceInfo( file_path=r.file_path, line=r.line, column=r.column, context=r.context, relationship_type=r.relationship_type, ) for r in refs ] # Get related symbols related_symbols = [] if include_related: related_symbols = self._get_related_symbols(symbol) return MCPContext( context_type=context_type, symbol=symbol_info, definition=definition, references=references, related_symbols=related_symbols, metadata={ "source": "codex-lens", "indexed_at": symbol.indexed_at if hasattr(symbol, 'indexed_at') else None, }, ) def _extract_definition(self, symbol) -> Optional[str]: """Extract source code for symbol definition.""" try: index_path = self.registry.find_index_path( Path(symbol.file_path).parent ) if not index_path: return None store = SQLiteStore(index_path.parent) rows = store.execute_query( "SELECT content FROM files WHERE full_path = ?", (symbol.file_path,), ) if not rows: return None content = rows[0]["content"] lines = content.split("\n") # Extract symbol lines start = symbol.range[0] - 1 end = symbol.range[1] return "\n".join(lines[start:end]) except Exception as e: logger.error(f"Failed to extract definition: {e}") return None def _get_related_symbols(self, symbol) -> List[RelatedSymbol]: """Get symbols related to the given symbol.""" related = [] try: index_path = self.registry.find_index_path( Path(symbol.file_path).parent ) if not index_path: return related store = SQLiteStore(index_path.parent) # Query relationships where this symbol is the source rows = store.execute_query( """ SELECT target_name, target_qualified_name, relationship_type FROM code_relationships WHERE source_qualified_name LIKE ? LIMIT 20 """, (f"%{symbol.name}%",), ) for row in rows: related.append(RelatedSymbol( name=row["target_name"], kind="unknown", # Would need another lookup relationship=row["relationship_type"], )) except Exception as e: logger.error(f"Failed to get related symbols: {e}") return related def build_context_for_file( self, file_path: Path, context_type: str = "file_overview", ) -> MCPContext: """Build context for an entire file.""" # Get all symbols in file symbols = self.global_index.search_by_file(str(file_path)) related = [ RelatedSymbol( name=s.name, kind=s.kind, relationship="defines", ) for s in symbols ] return MCPContext( context_type=context_type, related_symbols=related, metadata={ "file_path": str(file_path), "symbol_count": len(symbols), }, ) ``` --- #### Task 4.3: Hook Interfaces **File**: `src/codexlens/mcp/hooks.py` (NEW) ```python """Hook interfaces for Claude Code integration.""" import logging from pathlib import Path from typing import Any, Dict, Optional, Callable from codexlens.mcp.provider import MCPProvider from codexlens.mcp.schema import MCPContext logger = logging.getLogger(__name__) class HookManager: """Manages hook registration and execution.""" def __init__(self, mcp_provider: MCPProvider): self.mcp_provider = mcp_provider self._pre_hooks: Dict[str, Callable] = {} self._post_hooks: Dict[str, Callable] = {} # Register default hooks self._register_default_hooks() def _register_default_hooks(self): """Register built-in hooks.""" self._pre_hooks["explain"] = self._pre_explain_hook self._pre_hooks["refactor"] = self._pre_refactor_hook self._pre_hooks["document"] = self._pre_document_hook def execute_pre_hook( self, action: str, params: Dict[str, Any], ) -> Optional[MCPContext]: """Execute pre-tool hook to gather context. Args: action: The action being performed (e.g., "explain", "refactor") params: Parameters for the action Returns: MCPContext to inject into prompt, or None """ hook = self._pre_hooks.get(action) if not hook: logger.debug(f"No pre-hook for action: {action}") return None try: return hook(params) except Exception as e: logger.error(f"Pre-hook failed for {action}: {e}") return None def execute_post_hook( self, action: str, result: Any, ) -> None: """Execute post-tool hook for proactive caching. Args: action: The action that was performed result: Result of the action """ hook = self._post_hooks.get(action) if not hook: return try: hook(result) except Exception as e: logger.error(f"Post-hook failed for {action}: {e}") def _pre_explain_hook(self, params: Dict[str, Any]) -> Optional[MCPContext]: """Pre-hook for 'explain' action.""" symbol_name = params.get("symbol") if not symbol_name: return None return self.mcp_provider.build_context( symbol_name=symbol_name, context_type="symbol_explanation", include_references=True, include_related=True, ) def _pre_refactor_hook(self, params: Dict[str, Any]) -> Optional[MCPContext]: """Pre-hook for 'refactor' action.""" symbol_name = params.get("symbol") if not symbol_name: return None return self.mcp_provider.build_context( symbol_name=symbol_name, context_type="refactor_context", include_references=True, # Important for refactoring include_related=True, max_references=20, # More references for refactoring ) def _pre_document_hook(self, params: Dict[str, Any]) -> Optional[MCPContext]: """Pre-hook for 'document' action.""" symbol_name = params.get("symbol") file_path = params.get("file_path") if symbol_name: return self.mcp_provider.build_context( symbol_name=symbol_name, context_type="documentation_context", include_references=False, include_related=True, ) elif file_path: return self.mcp_provider.build_context_for_file( Path(file_path), context_type="file_documentation", ) return None def register_pre_hook( self, action: str, hook: Callable[[Dict[str, Any]], Optional[MCPContext]], ) -> None: """Register a custom pre-tool hook.""" self._pre_hooks[action] = hook def register_post_hook( self, action: str, hook: Callable[[Any], None], ) -> None: """Register a custom post-tool hook.""" self._post_hooks[action] = hook # Convenience function for Claude Code integration def create_context_for_prompt( mcp_provider: MCPProvider, action: str, params: Dict[str, Any], ) -> str: """Create context string for prompt injection. This is the main entry point for Claude Code hook integration. Args: mcp_provider: The MCP provider instance action: Action being performed params: Action parameters Returns: Formatted context string for prompt injection """ manager = HookManager(mcp_provider) context = manager.execute_pre_hook(action, params) if context: return context.to_prompt_injection() return "" ``` --- ## 8. Phase 5: Advanced Features ### 8.1 Custom LSP Commands **File**: `src/codexlens/lsp/handlers.py` (APPEND) ```python # Custom commands for advanced features @server.command("codexlens.hybridSearch") def cmd_hybrid_search(params: List[Any]) -> dict: """Execute hybrid search combining FTS and semantic.""" if len(params) < 1: return {"error": "Query required"} query = params[0] limit = params[1] if len(params) > 1 else 20 from codexlens.search.hybrid_search import HybridSearchEngine engine = HybridSearchEngine(server.search_engine.store) results = engine.search(query, limit=limit) return { "results": [ { "path": r.path, "score": r.score, "excerpt": r.excerpt, } for r in results ] } @server.command("codexlens.getMCPContext") def cmd_get_mcp_context(params: List[Any]) -> dict: """Get MCP context for a symbol.""" if len(params) < 1: return {"error": "Symbol name required"} symbol_name = params[0] context_type = params[1] if len(params) > 1 else "symbol_explanation" from codexlens.mcp.provider import MCPProvider provider = MCPProvider( server.global_index, server.search_engine, server.registry, ) context = provider.build_context(symbol_name, context_type) if context: return context.to_dict() return {"error": "Symbol not found"} ``` ### 8.2 Performance Optimizations **File**: `src/codexlens/lsp/cache.py` (NEW) ```python """Caching layer for LSP performance.""" import time from functools import lru_cache from typing import Any, Dict, Optional from threading import Lock class LRUCacheWithTTL: """LRU cache with time-to-live expiration.""" def __init__(self, maxsize: int = 1000, ttl_seconds: int = 300): self.maxsize = maxsize self.ttl = ttl_seconds self._cache: Dict[str, tuple] = {} # key -> (value, timestamp) self._lock = Lock() def get(self, key: str) -> Optional[Any]: """Get value from cache if not expired.""" with self._lock: if key not in self._cache: return None value, timestamp = self._cache[key] if time.time() - timestamp > self.ttl: del self._cache[key] return None return value def set(self, key: str, value: Any) -> None: """Set value in cache.""" with self._lock: # Evict oldest if at capacity if len(self._cache) >= self.maxsize: oldest_key = min( self._cache.keys(), key=lambda k: self._cache[k][1], ) del self._cache[oldest_key] self._cache[key] = (value, time.time()) def invalidate(self, key: str) -> None: """Remove key from cache.""" with self._lock: self._cache.pop(key, None) def invalidate_prefix(self, prefix: str) -> None: """Remove all keys with given prefix.""" with self._lock: keys_to_remove = [ k for k in self._cache.keys() if k.startswith(prefix) ] for key in keys_to_remove: del self._cache[key] def clear(self) -> None: """Clear all cache entries.""" with self._lock: self._cache.clear() # Global cache instances definition_cache = LRUCacheWithTTL(maxsize=500, ttl_seconds=300) references_cache = LRUCacheWithTTL(maxsize=200, ttl_seconds=60) completion_cache = LRUCacheWithTTL(maxsize=100, ttl_seconds=30) ``` --- ## 9. Testing Strategy ### 9.1 Test Structure ``` tests/ ├── lsp/ │ ├── __init__.py │ ├── conftest.py # Fixtures │ ├── test_server.py # Server lifecycle │ ├── test_definition.py # Definition handler │ ├── test_references.py # References handler │ ├── test_completion.py # Completion handler │ ├── test_hover.py # Hover handler │ └── test_workspace_symbol.py # Workspace symbol │ ├── mcp/ │ ├── __init__.py │ ├── test_schema.py # MCP schema validation │ ├── test_provider.py # Context building │ └── test_hooks.py # Hook execution │ └── integration/ ├── __init__.py ├── test_lsp_client.py # Full LSP handshake └── test_mcp_flow.py # End-to-end MCP flow ``` ### 9.2 Fixtures **File**: `tests/lsp/conftest.py` ```python """Test fixtures for LSP tests.""" import pytest from pathlib import Path import tempfile import shutil from codexlens.lsp.server import CodexLensLanguageServer @pytest.fixture def temp_workspace(): """Create temporary workspace with sample files.""" tmpdir = Path(tempfile.mkdtemp()) # Create sample Python files (tmpdir / "main.py").write_text(""" def main(): result = helper_function(42) print(result) def helper_function(x): return x * 2 """) (tmpdir / "utils.py").write_text(""" from main import helper_function class Calculator: def add(self, a, b): return a + b def multiply(self, a, b): return helper_function(a) * b """) yield tmpdir shutil.rmtree(tmpdir) @pytest.fixture def indexed_workspace(temp_workspace): """Workspace with built indexes.""" from codexlens.cli.commands import index_directory index_directory(temp_workspace) return temp_workspace @pytest.fixture def lsp_server(indexed_workspace): """Initialized LSP server.""" server = CodexLensLanguageServer("test", "0.1.0") server.initialize_codexlens(indexed_workspace) yield server server.shutdown_codexlens() ``` ### 9.3 Performance Benchmarks **File**: `tests/benchmarks/test_performance.py` ```python """Performance benchmarks for LSP operations.""" import pytest import time class TestPerformance: """Performance benchmark tests.""" @pytest.mark.benchmark def test_definition_latency(self, lsp_server, benchmark): """Definition lookup should be < 50ms.""" def lookup(): return lsp_server.global_index.search("helper_function", exact=True) result = benchmark(lookup) assert benchmark.stats.stats.mean < 0.05 # 50ms @pytest.mark.benchmark def test_completion_latency(self, lsp_server, benchmark): """Completion should be < 100ms.""" def complete(): return lsp_server.global_index.search("help", prefix_mode=True, limit=50) result = benchmark(complete) assert benchmark.stats.stats.mean < 0.1 # 100ms @pytest.mark.benchmark def test_references_latency(self, lsp_server, benchmark): """References should be < 200ms.""" def find_refs(): return lsp_server.search_engine.search_references("helper_function") result = benchmark(find_refs) assert benchmark.stats.stats.mean < 0.2 # 200ms ``` --- ## 10. Deployment Guide ### 10.1 Installation ```bash # Install with LSP support pip install codex-lens[lsp] # Or from source git clone https://github.com/your-org/codex-lens.git cd codex-lens pip install -e ".[lsp]" ``` ### 10.2 VS Code Configuration **File**: `.vscode/settings.json` ```json { "codexlens.enable": true, "codexlens.serverPath": "codexlens-lsp", "codexlens.serverArgs": ["--stdio"], "codexlens.trace.server": "verbose" } ``` ### 10.3 Neovim Configuration **File**: `~/.config/nvim/lua/lsp/codexlens.lua` ```lua local lspconfig = require('lspconfig') local configs = require('lspconfig.configs') configs.codexlens = { default_config = { cmd = { 'codexlens-lsp', '--stdio' }, filetypes = { 'python', 'javascript', 'typescript' }, root_dir = lspconfig.util.root_pattern('.git', 'pyproject.toml'), settings = {}, }, } lspconfig.codexlens.setup{} ``` ### 10.4 Claude Code Integration **File**: `~/.claude/hooks/pre-tool.sh` ```bash #!/bin/bash # Pre-tool hook for Claude Code ACTION="$1" PARAMS="$2" # Call codex-lens MCP provider python -c " from codexlens.mcp.hooks import create_context_for_prompt from codexlens.mcp.provider import MCPProvider from codexlens.storage.global_index import GlobalSymbolIndex from codexlens.search.chain_search import ChainSearchEngine from codexlens.storage.registry import RegistryStore from codexlens.storage.path_mapper import PathMapper import json # Initialize components registry = RegistryStore() registry.initialize() mapper = PathMapper() search = ChainSearchEngine(registry, mapper) global_idx = GlobalSymbolIndex(Path.cwd()) provider = MCPProvider(global_idx, search, registry) params = json.loads('$PARAMS') context = create_context_for_prompt(provider, '$ACTION', params) print(context) " ``` --- ## 11. Risk Mitigation ### 11.1 Risk Matrix | Risk | Probability | Impact | Mitigation | |------|-------------|--------|------------| | pygls compatibility issues | Low | High | Pin version, test on multiple platforms | | Performance degradation | Medium | Medium | Implement caching, benchmark tests | | Index corruption | Low | High | Use WAL mode, implement recovery | | Memory leaks in long sessions | Medium | Medium | Implement connection pooling, periodic cleanup | | Hook execution timeout | Medium | Low | Implement timeout limits, async execution | ### 11.2 Fallback Strategies 1. **Index not available**: Return empty results, don't block LSP 2. **Search timeout**: Return partial results with warning 3. **WatcherManager crash**: Auto-restart with exponential backoff 4. **MCP generation failure**: Return minimal context, log error ### 11.3 Monitoring ```python # Add to server.py import prometheus_client # Metrics DEFINITION_LATENCY = prometheus_client.Histogram( 'codexlens_definition_latency_seconds', 'Time to process definition request', ) REFERENCES_LATENCY = prometheus_client.Histogram( 'codexlens_references_latency_seconds', 'Time to process references request', ) INDEX_SIZE = prometheus_client.Gauge( 'codexlens_index_symbols_total', 'Total symbols in index', ) ``` --- ## Appendix: Quick Reference ### File Creation Summary | Phase | File | Type | |-------|------|------| | 1 | `src/codexlens/lsp/__init__.py` | NEW | | 1 | `src/codexlens/lsp/server.py` | NEW | | 1 | `src/codexlens/lsp/handlers.py` | NEW | | 2 | `src/codexlens/search/chain_search.py` | MODIFY | | 3 | `src/codexlens/lsp/providers.py` | NEW | | 4 | `src/codexlens/mcp/__init__.py` | NEW | | 4 | `src/codexlens/mcp/schema.py` | NEW | | 4 | `src/codexlens/mcp/provider.py` | NEW | | 4 | `src/codexlens/mcp/hooks.py` | NEW | | 5 | `src/codexlens/lsp/cache.py` | NEW | ### Command Reference ```bash # Start LSP server codexlens-lsp --stdio # Start with TCP (for debugging) codexlens-lsp --tcp --port 2087 # Run tests pytest tests/lsp/ -v # Run benchmarks pytest tests/benchmarks/ --benchmark-only # Check coverage pytest tests/lsp/ --cov=codexlens.lsp --cov-report=html ``` --- **Document End**