Migrate LSP client to vscode-jsonrpc for improved stability (#1095)

* refactor(lsp): migrate to vscode-jsonrpc for improved stability Replace custom JSON-RPC implementation with vscode-jsonrpc library. Use MessageConnection with StreamMessageReader/Writer. Implement Bun↔Node stream bridges for compatibility. Preserve all existing functionality (warmup, cleanup, capabilities). Net reduction of ~60 lines while improving protocol handling. * fix(lsp): clear timeout on successful response to prevent unhandled rejections --------- Co-authored-by: justsisyphus <justsisyphus@users.noreply.github.com>
2026-01-29 19:48:28 +09:00
parent 23b49c4a5c
commit a94fbadd57
3 changed files with 113 additions and 166 deletions
--- a/bun.lock
+++ b/bun.lock
@@ -18,6 +18,7 @@
        "jsonc-parser": "^3.3.1",
        "picocolors": "^1.1.1",
        "picomatch": "^4.0.2",
+        "vscode-jsonrpc": "^8.2.0",
        "zod": "^4.1.8",
      },
      "devDependencies": {
@@ -303,6 +304,8 @@

    "vary": ["vary@1.1.2", "", {}, "sha512-BNGbWLfd0eUPabhkXUVm0j8uuvREyTh5ovRa/dyow/BqAbZJyC+5fU+IzQOzmAKzYqYRAISoRhdQr3eIZ/PXqg=="],

+    "vscode-jsonrpc": ["vscode-jsonrpc@8.2.1", "", {}, "sha512-kdjOSJ2lLIn7r1rtrMbbNCHjyMPfRnowdKjBQ+mGq6NAW5QY2bEZC/khaC5OR8svbbjvLEaIXkOq45e2X9BIbQ=="],
+
    "which": ["which@2.0.2", "", { "dependencies": { "isexe": "^2.0.0" }, "bin": { "node-which": "./bin/node-which" } }, "sha512-BLI3Tl1TW3Pvl70l3yq3Y64i+awpwXqsGBYWkkqMtnbXgrMD+yj7rhW0kuEDxzJaYXGjEW5ogapKNMEKNMjibA=="],

    "wrappy": ["wrappy@1.0.2", "", {}, "sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ=="],
--- a/package.json
+++ b/package.json
@@ -64,6 +64,7 @@
    "jsonc-parser": "^3.3.1",
    "picocolors": "^1.1.1",
    "picomatch": "^4.0.2",
+    "vscode-jsonrpc": "^8.2.0",
    "zod": "^4.1.8"
  },
  "devDependencies": {
--- a/src/tools/lsp/client.ts
+++ b/src/tools/lsp/client.ts
@@ -1,7 +1,14 @@
 import { spawn, type Subprocess } from "bun"
+import { Readable, Writable } from "node:stream"
 import { readFileSync } from "fs"
 import { extname, resolve } from "path"
 import { pathToFileURL } from "node:url"
+import {
+  createMessageConnection,
+  StreamMessageReader,
+  StreamMessageWriter,
+  type MessageConnection,
+} from "vscode-jsonrpc/node"
 import { getLanguageId } from "./config"
 import type { Diagnostic, ResolvedServer } from "./types"

@@ -38,22 +45,18 @@ class LSPServerManager {
      }
    }

-    // Works on all platforms
    process.on("exit", cleanup)

-    // Ctrl+C - works on all platforms
    process.on("SIGINT", () => {
      cleanup()
      process.exit(0)
    })

-    // Kill signal - Unix/macOS
    process.on("SIGTERM", () => {
      cleanup()
      process.exit(0)
    })

-    // Ctrl+Break - Windows specific
    if (process.platform === "win32") {
      process.on("SIGBREAK", () => {
        cleanup()
@@ -209,13 +212,12 @@ export const lspManager = LSPServerManager.getInstance()

 export class LSPClient {
  private proc: Subprocess<"pipe", "pipe", "pipe"> | null = null
-  private buffer: Uint8Array = new Uint8Array(0)
-  private pending = new Map<number, { resolve: (value: unknown) => void; reject: (error: Error) => void }>()
-  private requestIdCounter = 0
+  private connection: MessageConnection | null = null
  private openedFiles = new Set<string>()
  private stderrBuffer: string[] = []
  private processExited = false
  private diagnosticsStore = new Map<string, Diagnostic[]>()
+  private readonly REQUEST_TIMEOUT = 15000

  constructor(
    private root: string,
@@ -238,7 +240,6 @@ export class LSPClient {
      throw new Error(`Failed to spawn LSP server: ${this.server.command.join(" ")}`)
    }

-    this.startReading()
    this.startStderrReading()

    await new Promise((resolve) => setTimeout(resolve, 100))
@@ -249,33 +250,66 @@ export class LSPClient {
        `LSP server exited immediately with code ${this.proc.exitCode}` + (stderr ? `\nstderr: ${stderr}` : "")
      )
    }
-  }

-  private startReading(): void {
-    if (!this.proc) return
-
-    const reader = this.proc.stdout.getReader()
-    const read = async () => {
-      try {
-        while (true) {
-          const { done, value } = await reader.read()
+    const stdoutReader = this.proc.stdout.getReader()
+    const nodeReadable = new Readable({
+      async read() {
+        try {
+          const { done, value } = await stdoutReader.read()
          if (done) {
-            this.processExited = true
-            this.rejectAllPending("LSP server stdout closed")
-            break
+            this.push(null)
+          } else {
+            this.push(Buffer.from(value))
          }
-          const newBuf = new Uint8Array(this.buffer.length + value.length)
-          newBuf.set(this.buffer)
-          newBuf.set(value, this.buffer.length)
-          this.buffer = newBuf
-          this.processBuffer()
+        } catch {
+          this.push(null)
        }
-      } catch (err) {
-        this.processExited = true
-        this.rejectAllPending(`LSP stdout read error: ${err}`)
+      },
+    })
+
+    const stdin = this.proc.stdin
+    const nodeWritable = new Writable({
+      write(chunk, _encoding, callback) {
+        try {
+          stdin.write(chunk)
+          callback()
+        } catch (err) {
+          callback(err as Error)
+        }
+      },
+    })
+
+    this.connection = createMessageConnection(
+      new StreamMessageReader(nodeReadable),
+      new StreamMessageWriter(nodeWritable)
+    )
+
+    this.connection.onNotification("textDocument/publishDiagnostics", (params: { uri?: string; diagnostics?: Diagnostic[] }) => {
+      if (params.uri) {
+        this.diagnosticsStore.set(params.uri, params.diagnostics ?? [])
      }
-    }
-    read()
+    })
+
+    this.connection.onRequest("workspace/configuration", (params: { items?: Array<{ section?: string }> }) => {
+      const items = params?.items ?? []
+      return items.map((item) => {
+        if (item.section === "json") return { validate: { enable: true } }
+        return {}
+      })
+    })
+
+    this.connection.onRequest("client/registerCapability", () => null)
+    this.connection.onRequest("window/workDoneProgress/create", () => null)
+
+    this.connection.onClose(() => {
+      this.processExited = true
+    })
+
+    this.connection.onError((error) => {
+      console.error("LSP connection error:", error)
+    })
+
+    this.connection.listen()
  }

  private startStderrReading(): void {
@@ -294,142 +328,49 @@ export class LSPClient {
            this.stderrBuffer.shift()
          }
        }
-      } catch {
-      }
+      } catch {}
    }
    read()
  }

-  private rejectAllPending(reason: string): void {
-    for (const [id, handler] of this.pending) {
-      handler.reject(new Error(reason))
-      this.pending.delete(id)
-    }
-  }
+  private async sendRequest<T>(method: string, params?: unknown): Promise<T> {
+    if (!this.connection) throw new Error("LSP client not started")

-  private findSequence(haystack: Uint8Array, needle: number[]): number {
-    outer: for (let i = 0; i <= haystack.length - needle.length; i++) {
-      for (let j = 0; j < needle.length; j++) {
-        if (haystack[i + j] !== needle[j]) continue outer
-      }
-      return i
-    }
-    return -1
-  }
-
-  private processBuffer(): void {
-    const decoder = new TextDecoder()
-    const CONTENT_LENGTH = [67, 111, 110, 116, 101, 110, 116, 45, 76, 101, 110, 103, 116, 104, 58]
-    const CRLF_CRLF = [13, 10, 13, 10]
-    const LF_LF = [10, 10]
-
-    while (true) {
-      const headerStart = this.findSequence(this.buffer, CONTENT_LENGTH)
-      if (headerStart === -1) break
-      if (headerStart > 0) this.buffer = this.buffer.slice(headerStart)
-
-      let headerEnd = this.findSequence(this.buffer, CRLF_CRLF)
-      let sepLen = 4
-      if (headerEnd === -1) {
-        headerEnd = this.findSequence(this.buffer, LF_LF)
-        sepLen = 2
-      }
-      if (headerEnd === -1) break
-
-      const header = decoder.decode(this.buffer.slice(0, headerEnd))
-      const match = header.match(/Content-Length:\s*(\d+)/i)
-      if (!match) break
-
-      const len = parseInt(match[1], 10)
-      const start = headerEnd + sepLen
-      const end = start + len
-      if (this.buffer.length < end) break
-
-      const content = decoder.decode(this.buffer.slice(start, end))
-      this.buffer = this.buffer.slice(end)
-
-      try {
-        const msg = JSON.parse(content)
-
-        if ("method" in msg && !("id" in msg)) {
-          if (msg.method === "textDocument/publishDiagnostics" && msg.params?.uri) {
-            this.diagnosticsStore.set(msg.params.uri, msg.params.diagnostics ?? [])
-          }
-        } else if ("id" in msg && "method" in msg) {
-          this.handleServerRequest(msg.id, msg.method, msg.params)
-        } else if ("id" in msg && this.pending.has(msg.id)) {
-          const handler = this.pending.get(msg.id)!
-          this.pending.delete(msg.id)
-          if ("error" in msg) {
-            handler.reject(new Error(msg.error.message))
-          } else {
-            handler.resolve(msg.result)
-          }
-        }
-      } catch {
-      }
-    }
-  }
-
-  private send(method: string, params?: unknown): Promise<unknown> {
-    if (!this.proc) throw new Error("LSP client not started")
-
-    if (this.processExited || this.proc.exitCode !== null) {
+    if (this.processExited || (this.proc && this.proc.exitCode !== null)) {
      const stderr = this.stderrBuffer.slice(-10).join("\n")
-      throw new Error(`LSP server already exited (code: ${this.proc.exitCode})` + (stderr ? `\nstderr: ${stderr}` : ""))
+      throw new Error(`LSP server already exited (code: ${this.proc?.exitCode})` + (stderr ? `\nstderr: ${stderr}` : ""))
    }

-    const id = ++this.requestIdCounter
-    const msg = JSON.stringify({ jsonrpc: "2.0", id, method, params })
-    const header = `Content-Length: ${Buffer.byteLength(msg)}\r\n\r\n`
-    this.proc.stdin.write(header + msg)
-
-    return new Promise((resolve, reject) => {
-      this.pending.set(id, { resolve, reject })
-      setTimeout(() => {
-        if (this.pending.has(id)) {
-          this.pending.delete(id)
-          const stderr = this.stderrBuffer.slice(-5).join("\n")
-          reject(new Error(`LSP request timeout (method: ${method})` + (stderr ? `\nrecent stderr: ${stderr}` : "")))
-        }
-      }, 15000)
+    let timeoutId: ReturnType<typeof setTimeout>
+    const timeoutPromise = new Promise<never>((_, reject) => {
+      timeoutId = setTimeout(() => {
+        const stderr = this.stderrBuffer.slice(-5).join("\n")
+        reject(new Error(`LSP request timeout (method: ${method})` + (stderr ? `\nrecent stderr: ${stderr}` : "")))
+      }, this.REQUEST_TIMEOUT)
    })
-  }

-  private notify(method: string, params?: unknown): void {
-    if (!this.proc) return
-    if (this.processExited || this.proc.exitCode !== null) return
+    const requestPromise = this.connection.sendRequest(method, params) as Promise<T>

-    const msg = JSON.stringify({ jsonrpc: "2.0", method, params })
-    this.proc.stdin.write(`Content-Length: ${Buffer.byteLength(msg)}\r\n\r\n${msg}`)
-  }
-
-  private respond(id: number | string, result: unknown): void {
-    if (!this.proc) return
-    if (this.processExited || this.proc.exitCode !== null) return
-
-    const msg = JSON.stringify({ jsonrpc: "2.0", id, result })
-    this.proc.stdin.write(`Content-Length: ${Buffer.byteLength(msg)}\r\n\r\n${msg}`)
-  }
-
-  private handleServerRequest(id: number | string, method: string, params?: unknown): void {
-    if (method === "workspace/configuration") {
-      const items = (params as { items?: Array<{ section?: string }> })?.items ?? []
-      const result = items.map((item) => {
-        if (item.section === "json") return { validate: { enable: true } }
-        return {}
-      })
-      this.respond(id, result)
-    } else if (method === "client/registerCapability") {
-      this.respond(id, null)
-    } else if (method === "window/workDoneProgress/create") {
-      this.respond(id, null)
+    try {
+      const result = await Promise.race([requestPromise, timeoutPromise])
+      clearTimeout(timeoutId!)
+      return result
+    } catch (error) {
+      clearTimeout(timeoutId!)
+      throw error
    }
  }

+  private sendNotification(method: string, params?: unknown): void {
+    if (!this.connection) return
+    if (this.processExited || (this.proc && this.proc.exitCode !== null)) return
+
+    this.connection.sendNotification(method, params)
+  }
+
  async initialize(): Promise<void> {
    const rootUri = pathToFileURL(this.root).href
-    await this.send("initialize", {
+    await this.sendRequest("initialize", {
      processId: process.pid,
      rootUri,
      rootPath: this.root,
@@ -481,8 +422,8 @@ export class LSPClient {
      },
      ...this.server.initialization,
    })
-    this.notify("initialized")
-    this.notify("workspace/didChangeConfiguration", {
+    this.sendNotification("initialized")
+    this.sendNotification("workspace/didChangeConfiguration", {
      settings: { json: { validate: { enable: true } } },
    })
    await new Promise((r) => setTimeout(r, 300))
@@ -496,7 +437,7 @@ export class LSPClient {
    const ext = extname(absPath)
    const languageId = getLanguageId(ext)

-    this.notify("textDocument/didOpen", {
+    this.sendNotification("textDocument/didOpen", {
      textDocument: {
        uri: pathToFileURL(absPath).href,
        languageId,
@@ -512,7 +453,7 @@ export class LSPClient {
  async definition(filePath: string, line: number, character: number): Promise<unknown> {
    const absPath = resolve(filePath)
    await this.openFile(absPath)
-    return this.send("textDocument/definition", {
+    return this.sendRequest("textDocument/definition", {
      textDocument: { uri: pathToFileURL(absPath).href },
      position: { line: line - 1, character },
    })
@@ -521,7 +462,7 @@ export class LSPClient {
  async references(filePath: string, line: number, character: number, includeDeclaration = true): Promise<unknown> {
    const absPath = resolve(filePath)
    await this.openFile(absPath)
-    return this.send("textDocument/references", {
+    return this.sendRequest("textDocument/references", {
      textDocument: { uri: pathToFileURL(absPath).href },
      position: { line: line - 1, character },
      context: { includeDeclaration },
@@ -531,13 +472,13 @@ export class LSPClient {
  async documentSymbols(filePath: string): Promise<unknown> {
    const absPath = resolve(filePath)
    await this.openFile(absPath)
-    return this.send("textDocument/documentSymbol", {
+    return this.sendRequest("textDocument/documentSymbol", {
      textDocument: { uri: pathToFileURL(absPath).href },
    })
  }

  async workspaceSymbols(query: string): Promise<unknown> {
-    return this.send("workspace/symbol", { query })
+    return this.sendRequest("workspace/symbol", { query })
  }

  async diagnostics(filePath: string): Promise<{ items: Diagnostic[] }> {
@@ -547,14 +488,13 @@ export class LSPClient {
    await new Promise((r) => setTimeout(r, 500))

    try {
-      const result = await this.send("textDocument/diagnostic", {
+      const result = await this.sendRequest<{ items?: Diagnostic[] }>("textDocument/diagnostic", {
        textDocument: { uri },
      })
      if (result && typeof result === "object" && "items" in result) {
        return result as { items: Diagnostic[] }
      }
-    } catch {
-    }
+    } catch {}

    return { items: this.diagnosticsStore.get(uri) ?? [] }
  }
@@ -562,7 +502,7 @@ export class LSPClient {
  async prepareRename(filePath: string, line: number, character: number): Promise<unknown> {
    const absPath = resolve(filePath)
    await this.openFile(absPath)
-    return this.send("textDocument/prepareRename", {
+    return this.sendRequest("textDocument/prepareRename", {
      textDocument: { uri: pathToFileURL(absPath).href },
      position: { line: line - 1, character },
    })
@@ -571,7 +511,7 @@ export class LSPClient {
  async rename(filePath: string, line: number, character: number, newName: string): Promise<unknown> {
    const absPath = resolve(filePath)
    await this.openFile(absPath)
-    return this.send("textDocument/rename", {
+    return this.sendRequest("textDocument/rename", {
      textDocument: { uri: pathToFileURL(absPath).href },
      position: { line: line - 1, character },
      newName,
@@ -583,10 +523,13 @@ export class LSPClient {
  }

  async stop(): Promise<void> {
-    try {
-      this.notify("shutdown", {})
-      this.notify("exit")
-    } catch {
+    if (this.connection) {
+      try {
+        this.sendNotification("shutdown", {})
+        this.sendNotification("exit")
+      } catch {}
+      this.connection.dispose()
+      this.connection = null
    }
    this.proc?.kill()
    this.proc = null