「语言服务器与 IDE 工具链」

// 客户端 -> 服务器: 初始化, 声明自己支持的能力
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "processId": 12345,
    "rootUri": "file:///home/dev/project",
    "capabilities": {
      "textDocument": {
        "completion": { "completionItem": { "snippetSupport": true } },
        "hover": { "contentFormat": ["markdown", "plaintext"] },
        "publishDiagnostics": { "relatedInformation": true }
      }
    }
  }
}

// 服务器 -> 客户端: 应答, 声明自己实现了哪些能力
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "capabilities": {
      "textDocumentSync": { "openClose": true, "change": 2 },
      "completionProvider": { "triggerCharacters": [".", ":"] },
      "definitionProvider": true,
      "renameProvider": { "prepareProvider": true },
      "hoverProvider": true
    },
    "serverInfo": { "name": "example-ls", "version": "0.1.0" }
  }
}

// 增量同步: 只报告变化的那一小段
{
  "jsonrpc": "2.0",
  "method": "textDocument/didChange",
  "params": {
    "textDocument": { "uri": "file:///p/src/main.rs", "version": 42 },
    "contentChanges": [
      {
        "range": {
          "start": { "line": 10, "character": 4 },
          "end":   { "line": 10, "character": 9 }
        },
        "text": "count"
      }
    ]
  }
}

def apply_change(text, change):
    """把 LSP 的 range 替换应用到本地文本副本."""
    lines = text.split("\n")
    s, e = change["range"]["start"], change["range"]["end"]
    new_line = lines[s["line"]][:s["character"]] + change["text"] \
             + lines[e["line"]][e["character"]:]
    lines[s["line"]:e["line"] + 1] = new_line.split("\n")
    return "\n".join(lines)

print(apply_change("let x = tol;\n", {"range": {"start": {"line": 0, "character": 8},
                                                "end": {"line": 0, "character": 11}},
                                      "text": "total"}))

class Node:
    __slots__ = ("kind", "start", "end", "children")
    def __init__(self, kind, start, end, children=()):
        self.kind, self.start, self.end = kind, start, end
        self.children = list(children)

def reuse_unchanged(node, change_start, change_end):
    """改动范围之外的子树可原样复用, 只重建路径上的节点."""
    if node.end <= change_start or node.start >= change_end:
        return node                      # 完全在改动范围外: 复用
    new_children = [reuse_unchanged(c, change_start, change_end)
                    for c in node.children]
    return Node(node.kind, node.start, node.end, new_children)

class SymbolIndex:
    def __init__(self):
        self.defs = {}          # 名字 -> [定义位置]
        self.refs = {}          # 名字 -> [引用位置]
        self.scopes = []        # 作用域栈: (名字, 定义位置)

    def define(self, name, loc, kind):
        self.defs.setdefault(name, []).append({"loc": loc, "kind": kind})
        self.scopes.append((name, loc))

    def resolve(self, name):
        """就近解析: 从作用域栈顶部向下找第一个匹配."""
        for n, loc in reversed(self.scopes):
            if n == name:
                return loc
        return None

idx = SymbolIndex()
idx.define("total", (3, 8), "variable")
idx.define("total", (9, 4), "parameter")     # 内层遮蔽
print("解析 total ->", idx.resolve("total"))

// 请求
{ "jsonrpc": "2.0", "id": 7, "method": "textDocument/definition",
  "params": { "textDocument": { "uri": "file:///p/main.py" },
              "position": { "line": 12, "character": 6 } } }
// 应答
{ "jsonrpc": "2.0", "id": 7,
  "result": { "uri": "file:///p/util.py",
              "range": { "start": { "line": 3, "character": 0 },
                         "end":   { "line": 3, "character": 9 } } } }

def fuzzy_score(prefix, name):
    """极简模糊匹配: 连续前缀给高分, 子序列给低分, 否则不匹配."""
    if not prefix: return 1
    if name.startswith(prefix): return 100 - len(name)
    it = iter(name.lower())
    return 10 if all(c in it for c in prefix.lower()) else 0

def complete_members(type_name, type_table, prefix=""):
    """成员补全: 按类型查成员表, 再做模糊过滤与排序."""
    scored = [(fuzzy_score(prefix, m["name"]), m)
              for m in type_table.get(type_name, [])]
    scored = [t for t in scored if t[0] > 0]
    scored.sort(key=lambda t: -t[0])
    return [m for _, m in scored][:50]      # 裁剪, 避免返回上万条

types = {"User": [{"name": "name"}, {"name": "email"}, {"name": "created_at"}]}
print(complete_members("User", types, "na"))

// 服务器推送的诊断对象: 位置 + 严重级别 + 来源 + 消息
{
  "range": { "start": { "line": 12, "character": 6 },
             "end":   { "line": 12, "character": 11 } },
  "severity": 1,
  "source": "example-ls",
  "message": "未定义的变量 total",
  "relatedInformation": []
}

from functools import lru_cache

@lru_cache(maxsize=4096)
def type_of_symbol(symbol_id, revision):
    """按 (符号, 文档版本) 缓存类型查询, 版本变化即自动失效."""
    return {"kind": "int"}             # 真实实现里是昂贵的类型推断

print(type_of_symbol(1, 42))
print(type_of_symbol.cache_info())     # hits/misses 可观测

class MemoTable:
    """带失效传播的记忆化: 依赖变化时只清理受影响的条目."""
    def __init__(self):
        self.cache, self.deps = {}, {}

    def get(self, key, compute):
        if key not in self.cache:
            self.cache[key] = compute()
        return self.cache[key]

    def invalidate(self, changed):
        for k, deps in list(self.deps.items()):
            if changed & deps:
                self.cache.pop(k, None)
                self.deps.pop(k, None)

import json, sys

DOCS = {}

def read_message(stream):
    length = 0
    while (line := stream.readline()) not in (b"\r\n", b"\n", b""):
        if line.lower().startswith(b"content-length:"):
            length = int(line.split(b":")[1])
    return json.loads(stream.read(length)) if length else None

def write_message(msg):
    body = json.dumps(msg).encode("utf-8")
    sys.stdout.buffer.write(f"Content-Length: {len(body)}\r\n\r\n".encode())
    sys.stdout.buffer.write(body)
    sys.stdout.buffer.flush()

def handle(msg):
    method = msg.get("method")
    if method == "initialize":
        return {"capabilities": {"hoverProvider": True,
                                 "completionProvider": {"triggerCharacters": ["."]},
                                 "textDocumentSync": 1}}
    if method == "textDocument/didOpen":
        uri = msg["params"]["textDocument"]["uri"]
        DOCS[uri] = msg["params"]["textDocument"]["text"]
        return None
    if method == "textDocument/hover":
        return {"contents": {"kind": "markdown", "value": "**example-ls** 悬停提示"}}
    if method == "textDocument/completion":
        return {"isIncomplete": False,
                "items": [{"label": "total"}, {"label": "count"}]}
    return None

def main():
    while True:
        msg = read_message(sys.stdin.buffer)
        if msg is None:
            break
        result = handle(msg)
        if "id" in msg:
            write_message({"jsonrpc": "2.0", "id": msg["id"], "result": result})

# 推送诊断: 服务器主动通知客户端
def publish_diagnostics(uri, text):
    problems = []
    for i, line in enumerate(text.split("\n")):
        if line.count("(") != line.count(")"):
            problems.append({
                "range": {"start": {"line": i, "character": 0},
                          "end": {"line": i, "character": len(line)}},
                "severity": 1, "source": "example-ls",
                "message": "括号不匹配",
            })
    write_message({"jsonrpc": "2.0", "method": "textDocument/publishDiagnostics",
                   "params": {"uri": uri, "diagnostics": problems}})