feat: 自研本地 LSP 服务器 (jedi 语义 + py_detect 诊断)，替代 pylsp

- xcli_core/local_lsp_server.py: LSP over stdio (Content-Length 分帧)，兼容现有 lsp_client - 语义: jedi 0.20 API (goto/infer/get_references/complete/help) — 补全/跳转/引用/悬停 - 诊断: 复用 plugins/py_detect 规则引擎，中文诊断 + 修复建议，didOpen/didChange 自动推送 - 零网络、零外部安装 (无需 pip install pylsp)，服务器无响应时客户端可超时跳过 - config.json(.example): python 语言默认切到本地服务器; 换回 pylsp 改 command 即可 - lsp_client.py: command 为裸 python 且 PATH 找不到时回退 sys.executable (防环境坑) - 16 项测试 (分帧/路径/协议/语义/真实 subprocess 完整握手); 相关套件 277 passed 零回归
2026-08-07 02:00:49 +08:00
1 parent 9a133dcf88
commit 56d5ed27e0
@@ -19,10 +19,10 @@
    "max_history": 50
  },
  "lspServers": {
-    "_说明": "语言服务器（LSP）。键为语言ID或扩展名，值为启动命令。配置后重启即生效，工具 lsp__diagnostics/hover/definition/references/completion 自动可用，且 AI 写/改 .py 文件后会自动触发诊断并注入模型上下文。",
+    "_说明": "语言服务器（LSP）。键为语言ID或扩展名，值为启动命令。配置后重启即生效，工具 lsp__diagnostics/hover/definition/references/completion 自动可用，且 AI 写/改 .py 文件后会自动触发诊断并注入模型上下文。python 默认用自研本地服务器（jedi 语义 + py_detect 中文诊断，纯本地零安装）；换回 pylsp 则 command 改 \"pylsp\"、args 清空。",
    "python": {
-      "command": "pylsp",
-      "args": [],
+      "command": "python",
+      "args": ["-m", "xcli_core.local_lsp_server"],
      "disabled": false
    }
  }
@@ -18,6 +18,8 @@ import os
 import re
 import json
 import time
+import shutil
+import sys
 import threading
 import subprocess
 import logging
@@ -115,6 +117,14 @@ class LspClient:
        full_env = dict(os.environ)
        full_env.update(self.env)
        cmd = [self.command] + self.args
+        # 鲁棒性: command 为裸 "python"/"python3" 但 PATH 找不到时, 回退当前解释器
+        if cmd and os.path.basename(cmd[0]) in ("python", "python3") and os.path.dirname(cmd[0]) == "":
+            found = None
+            for cand in (cmd[0], cmd[0] + ".exe"):
+                if found is None and shutil.which(cand):
+                    found = cand
+            if not found:
+                cmd = [sys.executable] + self.args
        try:
            # 二进制模式：避免 CJK 字符数与 Content-Length 字节数不一致
            self.proc = subprocess.Popen(
@@ -0,0 +1,229 @@
+"""自研本地 LSP 服务器测试：单元（分帧/协议/引擎）+ 真实 subprocess 集成"""
+import io
+import json
+import os
+import queue
+import re
+import subprocess
+import sys
+import threading
+
+import pytest
+
+PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, PROJECT_ROOT)
+
+from xcli_core.local_lsp_server import (
+    LocalLspServer,
+    path_to_uri,
+    read_message,
+    uri_to_path,
+    write_message,
+)
+
+SRC = "import math\ndef add(a, b):\n    return a + b\n\nresult = add(1, 2)\nprint(result)\n"
+URI = "file:///C:/Users/Administrator/Desktop/xiaoli-cli/_test_tmp.py"
+
+
+# ── 分帧传输层 ──
+class TestFraming:
+    def test_roundtrip(self):
+        buf = io.BytesIO()
+        write_message(buf, {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}})
+        buf.seek(0)
+        msg = read_message(buf)
+        assert msg == {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}
+
+    def test_cjk_bytes(self):
+        # 中文消息体: Content-Length 按字节计
+        obj = {"result": "未定义变量 — 中文诊断"}
+        buf = io.BytesIO()
+        write_message(buf, obj)
+        buf.seek(0)
+        assert read_message(buf) == obj
+
+    def test_empty_stream(self):
+        assert read_message(io.BytesIO(b"")) is None
+
+
+# ── 路径工具 ──
+class TestPath:
+    def test_uri_to_path_windows(self):
+        p = uri_to_path("file:///C:/Users/a/b.py")
+        assert os.path.abspath(p) == os.path.abspath("C:/Users/a/b.py")
+
+    def test_path_to_uri(self):
+        assert path_to_uri(r"C:\Users\a\b.py").startswith("file:///C:/Users/a/b.py")
+
+
+# ── 协议层（进程内直接调 handle） ──
+class TestProtocol:
+    def make_server(self):
+        srv = LocalLspServer()
+        srv.handle({"jsonrpc": "2.0", "method": "initialize", "id": 1,
+                    "params": {"rootUri": "file:///" + PROJECT_ROOT.replace("\\", "/")}})
+        return srv
+
+    def test_initialize_capabilities(self):
+        srv = LocalLspServer()
+        outs = srv.handle({"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}})
+        assert outs[0]["id"] == 1
+        caps = outs[0]["result"]["capabilities"]
+        assert caps["definitionProvider"] is True
+        assert "completionProvider" in caps
+        assert caps["textDocumentSync"]["change"] == 1
+
+    def test_unknown_method_error(self):
+        srv = LocalLspServer()
+        outs = srv.handle({"jsonrpc": "2.0", "method": "textDocument/xxx", "id": 9, "params": {}})
+        assert outs[0]["error"]["code"] == -32601
+
+    def test_shutdown_and_unknown_notification_no_reply(self):
+        srv = LocalLspServer()
+        assert srv.handle({"jsonrpc": "2.0", "method": "initialized", "params": {}}) == []
+        assert srv.handle({"jsonrpc": "2.0", "method": "shutdown", "id": 2,
+                           "params": {}})[0]["result"] is None
+
+    def test_did_open_pushes_diagnostics(self):
+        srv = self.make_server()
+        outs = srv.handle({"jsonrpc": "2.0", "method": "textDocument/didOpen", "params": {
+            "textDocument": {"uri": URI, "languageId": "python", "version": 1, "text": SRC}}})
+        push = [o for o in outs if o.get("method") == "textDocument/publishDiagnostics"]
+        assert push, "didOpen 应推送诊断"
+        diags = push[0]["params"]["diagnostics"]
+        assert diags, "应检测到问题 (import math 未使用)"
+        first = diags[0]
+        assert first["source"] == "py_detect"
+        assert "range" in first and "severity" in first and "message" in first
+
+    def test_diagnostics_clean_file_no_push(self):
+        srv = self.make_server()
+        clean = "def add(a, b):\n    return a + b\nprint(add(1, 2))\n"
+        outs = srv.handle({"jsonrpc": "2.0", "method": "textDocument/didOpen", "params": {
+            "textDocument": {"uri": URI, "languageId": "python", "version": 1, "text": clean}}})
+        assert not [o for o in outs if o.get("method") == "textDocument/publishDiagnostics"]
+
+
+# ── jedi 语义（进程内，无需真实文件） ──
+class TestSemantic:
+    def open(self):
+        srv = LocalLspServer()
+        srv.handle({"jsonrpc": "2.0", "method": "textDocument/didOpen", "params": {
+            "textDocument": {"uri": URI, "languageId": "python", "version": 1, "text": SRC}}})
+        return srv
+
+    def req(self, srv, method, line, character):
+        outs = srv.handle({"jsonrpc": "2.0", "method": method, "id": 1, "params": {
+            "textDocument": {"uri": URI}, "position": {"line": line, "character": character}}})
+        return outs[0]["result"]
+
+    def test_completion(self):
+        srv = self.open()
+        res = self.req(srv, "textDocument/completion", 6, 10)  # result.
+        items = res.get("items", []) if isinstance(res, dict) else res
+        labels = [i["label"] for i in items]
+        assert "result" in labels, f"应在 result 后补全: {labels}"
+
+    def test_definition(self):
+        srv = self.open()
+        res = self.req(srv, "textDocument/definition", 5, 10)  # add 调用处
+        assert res, "应找到 add 定义"
+        loc = res[0]
+        assert loc["uri"].startswith("file:///")
+        assert loc["range"]["start"]["line"] == 2  # 定义在第 3 行 (0-based)
+
+    def test_references(self):
+        srv = self.open()
+        res = self.req(srv, "textDocument/references", 5, 10)  # add
+        assert len(res) >= 2, "应至少含定义处+调用处"
+
+    def test_hover(self):
+        srv = self.open()
+        res = self.req(srv, "textDocument/hover", 5, 10)  # add
+        assert res is not None
+        contents = res.get("contents", {})
+        value = contents.get("value", "") if isinstance(contents, dict) else str(contents)
+        assert value, "hover 应有内容"
+
+    def test_hover_builtin(self):
+        srv = self.open()
+        # math 已 import; 悬停 print 是内置, 不崩溃即可
+        res = self.req(srv, "textDocument/hover", 6, 1)
+        assert res is None or isinstance(res, dict)
+
+
+# ── 真实 subprocess 集成（完整握手） ──
+class TestSubprocessIntegration:
+    def test_full_handshake(self):
+        proc = subprocess.Popen(
+            [sys.executable, "-m", "xcli_core.local_lsp_server"],
+            stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+            stderr=subprocess.DEVNULL, cwd=PROJECT_ROOT, bufsize=0)
+        try:
+            q = queue.Queue()
+
+            def _reader():
+                try:
+                    while True:
+                        header = b""
+                        while b"\r\n\r\n" not in header:
+                            chunk = proc.stdout.read(1)
+                            if not chunk:
+                                return
+                            header += chunk
+                        m = re.search(rb"Content-Length:\s*(\d+)", header)
+                        body = proc.stdout.read(int(m.group(1))) if m else b""
+                        if body:
+                            q.put(json.loads(body.decode("utf-8")))
+                except Exception:
+                    pass
+
+            threading.Thread(target=_reader, daemon=True).start()
+
+            def send(obj):
+                data = json.dumps(obj, ensure_ascii=False).encode("utf-8")
+                proc.stdin.write(f"Content-Length: {len(data)}\r\n\r\n".encode() + data)
+                proc.stdin.flush()
+
+            def recv(timeout=25):
+                return q.get(timeout=timeout)
+
+            root_uri = "file:///" + PROJECT_ROOT.replace("\\", "/")
+            send({"jsonrpc": "2.0", "id": 1, "method": "initialize",
+                  "params": {"processId": os.getpid(), "rootUri": root_uri,
+                             "capabilities": {"textDocument": {}}}})
+            r = recv()
+            assert "capabilities" in r.get("result", {}), r
+
+            send({"jsonrpc": "2.0", "method": "initialized", "params": {}})
+            send({"jsonrpc": "2.0", "method": "textDocument/didOpen", "params": {
+                "textDocument": {"uri": URI, "languageId": "python", "version": 1,
+                                 "text": SRC}}})
+            diag = recv()
+            assert diag.get("method") == "textDocument/publishDiagnostics", diag
+            assert diag["params"]["diagnostics"], "应推送诊断"
+
+            send({"jsonrpc": "2.0", "id": 2, "method": "textDocument/completion", "params": {
+                "textDocument": {"uri": URI}, "position": {"line": 6, "character": 10}}})
+            c = recv()["result"]
+            items = c.get("items", []) if isinstance(c, dict) else c
+            assert any(i["label"] == "result" for i in items)
+
+            send({"jsonrpc": "2.0", "id": 3, "method": "textDocument/definition", "params": {
+                "textDocument": {"uri": URI}, "position": {"line": 5, "character": 10}}})
+            d = recv()["result"]
+            assert d and d[0]["range"]["start"]["line"] == 2
+
+            send({"jsonrpc": "2.0", "id": 4, "method": "textDocument/hover", "params": {
+                "textDocument": {"uri": URI}, "position": {"line": 5, "character": 10}}})
+            h = recv()["result"]
+            assert h is not None
+
+            send({"jsonrpc": "2.0", "id": 5, "method": "shutdown", "params": {}})
+            assert recv()["result"] is None
+            send({"jsonrpc": "2.0", "method": "exit", "params": {}})
+            proc.stdin.close()
+            proc.wait(timeout=5)
+        finally:
+            if proc.poll() is None:
+                proc.kill()
@@ -0,0 +1,433 @@
+"""
+自研本地 LSP 服务器 (local_lsp_server)
+======================================
+纯本地、零外部服务依赖的 Python 语言服务器——替代 pylsp。
+
+架构:
+  - 语义引擎: jedi (纯 Python, 已在环境) — 补全/跳转/引用/悬停/签名
+  - 诊断引擎: plugins/py_detect (ast 规则引擎, 中文诊断)
+  - 协议: LSP over stdio (Content-Length 分帧), 与 xiaoli 的 lsp_client.py 完全兼容
+
+特性:
+  - 零网络请求, 不依赖任何外部安装 (无需 pip install pylsp)
+  - 服务器不响应时客户端可超时跳过, 不会无限卡死
+  - 诊断推送自动携带 py_detect 的中文解释 + 修复建议
+
+用法:
+  python -m xcli_core.local_lsp_server
+  或配置进 config.json 的 lspServers: {"command": "python", "args": ["-m", "xcli_core.local_lsp_server"]}
+
+支持的 LSP 方法:
+  initialize / initialized / shutdown / exit
+  textDocument/didOpen|didChange|didClose (通知)
+  textDocument/completion|definition|references|hover (请求)
+  textDocument/publishDiagnostics (主动推送, 来自 py_detect)
+"""
+
+import json
+import os
+import re
+import sys
+from typing import Any, Dict, List, Optional
+
+__version__ = "1.0.0"
+
+# ══════════════════════════════════════════════
+#  Content-Length 分帧 (LSP stdio 传输层)
+# ══════════════════════════════════════════════
+
+
+def read_message(stream) -> Optional[Dict[str, Any]]:
+    """从流中读取一条 Content-Length 分帧消息"""
+    header = b""
+    while b"\r\n\r\n" not in header:
+        chunk = stream.read(1)
+        if not chunk:
+            return None
+        header += chunk
+        if len(header) > 65536:
+            return None
+    m = re.search(rb"Content-Length:\s*(\d+)", header)
+    if not m:
+        return None
+    length = int(m.group(1))
+    body = b""
+    while len(body) < length:
+        chunk = stream.read(length - len(body))
+        if not chunk:
+            return None
+        body += chunk
+    try:
+        return json.loads(body.decode("utf-8"))
+    except (json.JSONDecodeError, UnicodeDecodeError):
+        return None
+
+
+def write_message(stream, obj: Dict[str, Any]) -> None:
+    """向流中写入一条 Content-Length 分帧消息"""
+    data = json.dumps(obj, ensure_ascii=False).encode("utf-8")
+    stream.write(f"Content-Length: {len(data)}\r\n\r\n".encode("ascii"))
+    stream.write(data)
+    stream.flush()
+
+
+# ══════════════════════════════════════════════
+#  路径工具 (file:// URI <-> 文件系统路径)
+# ══════════════════════════════════════════════
+
+
+def uri_to_path(uri: str) -> str:
+    """file:///C:/a/b.py -> C:/a/b.py"""
+    if uri.startswith("file://"):
+        path = uri[7:]
+        if len(path) >= 3 and path[0] == "/" and path[2] == ":":
+            path = path[1:]  # /C:/... -> C:/...
+        return path.replace("/", os.sep) if os.sep == "\\" else path
+    return uri
+
+
+def path_to_uri(path: str) -> str:
+    abspath = os.path.abspath(path)
+    norm = abspath.replace("\\", "/")
+    if not norm.startswith("/"):
+        norm = "/" + norm
+    return "file://" + norm
+
+
+# ══════════════════════════════════════════════
+#  自研本地 LSP 服务器
+# ══════════════════════════════════════════════
+
+
+class LocalLspServer:
+    """本地 Python 语言服务器: jedi 语义 + py_detect 诊断。
+
+    handle(msg) 返回要写出的消息列表（响应/推送），便于单元测试与进程模式复用。
+    """
+
+    def __init__(self):
+        self.documents: Dict[str, str] = {}      # uri -> 最新文本
+        self.root_path: Optional[str] = None
+        self._project = None
+        self._jedi = None
+        self._py_detect = None
+        self._shutdown = False
+
+    # ── 引擎惰性加载 (首次用时才 import, 失败不崩) ──
+    def _get_jedi(self):
+        if self._jedi is None:
+            try:
+                import jedi
+                self._jedi = jedi
+            except Exception:
+                self._jedi = False
+        return self._jedi or None
+
+    def _get_project(self):
+        if self._project is None and self._get_jedi() is not None:
+            try:
+                root = self.root_path
+                if root and os.path.isdir(root):
+                    self._project = self._jedi.Project(root)
+                else:
+                    self._project = False
+            except Exception:
+                self._project = False
+        return self._project if self._project else None
+
+    def _load_py_detect(self):
+        """从 plugins/py_detect.py 加载诊断引擎 (与 code_editor 同源, 零复制)"""
+        if self._py_detect is not None:
+            return self._py_detect if self._py_detect is not False else None
+        try:
+            import importlib.util
+            here = os.path.dirname(os.path.abspath(__file__))
+            plugin_path = os.path.join(here, "..", "plugins", "py_detect.py")
+            if not os.path.exists(plugin_path):
+                self._py_detect = False
+                return None
+            spec = importlib.util.spec_from_file_location("py_detect_server", plugin_path)
+            mod = importlib.util.module_from_spec(spec)
+            spec.loader.exec_module(mod)
+            self._py_detect = mod
+            return mod
+        except Exception:
+            self._py_detect = False
+            return None
+
+    def _script(self, uri: str, text: Optional[str] = None):
+        jedi = self._get_jedi()
+        if jedi is None:
+            return None
+        path = uri_to_path(uri)
+        code = text if text is not None else self.documents.get(uri, "")
+        return jedi.Script(code=code, path=path, project=self._get_project())
+
+    # ── 消息分发: 返回要写出的消息列表 ──
+    def handle(self, msg: Dict[str, Any]) -> List[Dict[str, Any]]:
+        method = msg.get("method")
+        msg_id = msg.get("id")
+        params = msg.get("params") or {}
+
+        # 客户端不应向服务器发响应; 防御性忽略
+        if method is None and msg_id is not None:
+            return []
+
+        if method == "initialize":
+            root_uri = params.get("rootUri") or ""
+            if root_uri:
+                self.root_path = uri_to_path(root_uri)
+            return [self._reply(msg_id, {"capabilities": self._capabilities()})]
+
+        if method == "initialized":
+            return []
+
+        if method == "shutdown":
+            self._shutdown = True
+            return [self._reply(msg_id, None)]
+
+        if method == "exit":
+            self._shutdown = True
+            return []  # 进程层据此退出
+
+        if method == "textDocument/didOpen":
+            td = params.get("textDocument", {})
+            uri = td.get("uri", "")
+            self.documents[uri] = td.get("text", "")
+            return self._diag_push(uri)
+
+        if method == "textDocument/didChange":
+            uri = (params.get("textDocument", {}) or {}).get("uri", "")
+            changes = params.get("contentChanges") or []
+            if changes:
+                self.documents[uri] = changes[-1].get("text", "")
+            return self._diag_push(uri)
+
+        if method == "textDocument/didClose":
+            uri = (params.get("textDocument", {}) or {}).get("uri", "")
+            self.documents.pop(uri, None)
+            return []
+
+        if method == "textDocument/completion":
+            return [self._reply(msg_id, self._completion(params))]
+
+        if method == "textDocument/definition":
+            return [self._reply(msg_id, self._definition(params))]
+
+        if method == "textDocument/references":
+            return [self._reply(msg_id, self._references(params))]
+
+        if method == "textDocument/hover":
+            return [self._reply(msg_id, self._hover(params))]
+
+        if msg_id is not None:
+            return [self._reply(msg_id, None,
+                                error={"code": -32601, "message": f"未知方法: {method}"})]
+        return []
+
+    def _reply(self, msg_id, result, error=None) -> Dict[str, Any]:
+        msg = {"jsonrpc": "2.0", "id": msg_id}
+        if error is not None:
+            msg["error"] = error
+        else:
+            msg["result"] = result
+        return msg
+
+    def _capabilities(self) -> Dict[str, Any]:
+        return {
+            "textDocumentSync": {"openClose": True, "change": 1},  # 1 = full
+            "completionProvider": {"triggerCharacters": [".", "(", ","]},
+            "definitionProvider": True,
+            "referencesProvider": True,
+            "hoverProvider": True,
+        }
+
+    # ── py_detect 诊断 → publishDiagnostics 推送 ──
+    def _diag_push(self, uri: str) -> List[Dict[str, Any]]:
+        pd = self._load_py_detect()
+        text = self.documents.get(uri, "")
+        diags = []
+        if pd is not None:
+            try:
+                issues = pd.analyze(text, uri_to_path(uri) or "<buffer>")
+                sev_map = {"错误": 1, "警告": 2, "提示": 4, "信息": 3}
+                for d in issues:
+                    line0 = max(int(d.get("line", 1)) - 1, 0)
+                    col0 = max(int(d.get("col", 1)) - 1, 0)
+                    sev = sev_map.get(d.get("severity", "警告"), 2)
+                    message = "[{}] {}{}".format(
+                        d.get("code", "?"),
+                        d.get("title", ""),
+                        (" — " + d.get("fix", "")) if d.get("fix") else "",
+                    )
+                    diags.append({
+                        "range": {
+                            "start": {"line": line0, "character": col0},
+                            "end": {"line": line0, "character": col0},
+                        },
+                        "severity": sev,
+                        "source": "py_detect",
+                        "message": message,
+                    })
+            except Exception:
+                diags = []
+        if not diags:
+            return []
+        return [{
+            "jsonrpc": "2.0",
+            "method": "textDocument/publishDiagnostics",
+            "params": {"uri": uri, "diagnostics": diags},
+        }]
+
+    # ── jedi 语义能力 ──
+    def _position(self, params: Dict[str, Any]):
+        pos = params.get("position") or {}
+        return int(pos.get("line", 0) or 0), int(pos.get("character", 0) or 0)
+
+    def _uri(self, params: Dict[str, Any]) -> str:
+        return (params.get("textDocument") or {}).get("uri", "")
+
+    def _completion(self, params: Dict[str, Any]) -> Dict[str, Any]:
+        uri = self._uri(params)
+        line, character = self._position(params)
+        script = self._script(uri)
+        if script is None:
+            return {"items": []}
+        try:
+            completions = script.complete(line, character)
+        except Exception:
+            return {"items": []}
+        items = []
+        seen = set()
+        for c in completions:
+            name = getattr(c, "name", "") or ""
+            if not name or name in seen:
+                continue
+            seen.add(name)
+            ctype = getattr(c, "type", "") or ""
+            detail = (getattr(c, "description", "") or "")[:120]
+            items.append({
+                "label": name,
+                "kind": ctype,
+                "detail": detail,
+                "sortText": f"0{ctype}_{name}",
+            })
+        return {"items": items[:60]}
+
+    def _locations(self, defs) -> List[Dict[str, Any]]:
+        out = []
+        for d in defs:
+            module_path = getattr(d, "module_path", None)
+            if module_path is None:
+                continue  # 内置/无源码定义跳过
+            line = getattr(d, "line", 0) or 0
+            col = getattr(d, "column", 0) or 0
+            out.append({
+                "uri": path_to_uri(str(module_path)),
+                "range": {
+                    "start": {"line": line, "character": col},
+                    "end": {"line": line, "character": col},
+                },
+            })
+        return out
+
+    def _definition(self, params: Dict[str, Any]) -> List[Dict[str, Any]]:
+        uri = self._uri(params)
+        line, character = self._position(params)
+        script = self._script(uri)
+        if script is None:
+            return []
+        try:
+            defs = script.goto(line, character)  # jedi 0.20 API
+            return self._locations(defs)
+        except Exception:
+            return []
+
+    def _references(self, params: Dict[str, Any]) -> List[Dict[str, Any]]:
+        uri = self._uri(params)
+        line, character = self._position(params)
+        script = self._script(uri)
+        if script is None:
+            return []
+        try:
+            refs = script.get_references(line, character)  # jedi 0.20 API
+            return self._locations(refs)
+        except Exception:
+            return []
+
+    def _hover(self, params: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+        uri = self._uri(params)
+        line, character = self._position(params)
+        script = self._script(uri)
+        if script is None:
+            return None
+        parts = []
+        # 类型推断 (jedi 0.20: infer 替代 goto_definitions)
+        try:
+            inferred = script.infer(line, character)
+            seen_types = set()
+            for d in inferred[:5]:
+                dname = getattr(d, "name", "")
+                dtype = getattr(d, "type", "")
+                key = (dname, dtype)
+                if key in seen_types:
+                    continue
+                seen_types.add(key)
+                if dtype:
+                    parts.append(f"类型: {dname}: {dtype}")
+        except Exception:
+            pass
+        # 签名
+        try:
+            for s in script.get_signatures(line, character)[:3]:
+                stxt = getattr(s, "to_string", lambda: "")()
+                if stxt:
+                    parts.append(f"签名: {stxt}")
+        except Exception:
+            pass
+        # 文档
+        try:
+            for h in script.help(line, character)[:3]:
+                doc = getattr(h, "docstring", None)
+                if doc:
+                    d = doc() or ""
+                    if d:
+                        parts.append(d.strip()[:400])
+        except Exception:
+            pass
+        if not parts:
+            return None
+        return {"contents": {"kind": "plaintext", "value": "\n\n".join(parts[:6])}}
+
+
+# ══════════════════════════════════════════════
+#  进程入口
+# ══════════════════════════════════════════════
+
+
+def main() -> None:
+    server = LocalLspServer()
+    stdin = sys.stdin.buffer
+    stdout = sys.stdout.buffer
+    while True:
+        try:
+            msg = read_message(stdin)
+        except Exception:
+            break
+        if msg is None:
+            break
+        try:
+            outputs = server.handle(msg)
+        except Exception:
+            outputs = []
+        for out in outputs:
+            try:
+                write_message(stdout, out)
+            except (BrokenPipeError, OSError):
+                return
+        if server._shutdown and msg.get("method") == "exit":
+            break
+
+
+if __name__ == "__main__":
+    main()