Merge branch 'dev' into fix/lsp-dead-root-prune

packages/opencode/src/lsp/index.ts

@@ -2,6 +2,7 @@ import { BusEvent } from "@/bus/bus-event"
 import { Bus } from "@/bus"
 import { Log } from "../util/log"
 import { LSPClient } from "./client"
+import { watch as fswatch, type FSWatcher } from "fs"
 import path from "path"
 import { pathToFileURL, fileURLToPath } from "url"
 import { LSPServer } from "./server"
@@ -11,8 +12,9 @@ import { Instance } from "../project/instance"
 import { Flag } from "@/flag/flag"
 import { Process } from "../util/process"
 import { spawn as lspspawn } from "./launch"
-import { Effect, Layer, Context } from "effect"
+import { Effect, Layer, Context, PubSub, Stream } from "effect"
 import { InstanceState } from "@/effect/instance-state"
+import { Filesystem } from "@/util/filesystem"
 
 export namespace LSP {
   const log = Log.create({ service: "lsp" })
@@ -116,6 +118,8 @@ export namespace LSP {
     SymbolKind.Enum,
   ]
 
+  const key = (id: string, root: string) => `${id}\0${root}`
+
   const filterExperimentalServers = (servers: Record<string, LSPServer.Info>) => {
     if (Flag.OPENCODE_EXPERIMENTAL_LSP_TY) {
       if (servers["pyright"]) {
@@ -135,7 +139,10 @@ export namespace LSP {
     clients: LSPClient.Info[]
     servers: Record<string, LSPServer.Info>
     broken: Set<string>
+    pulse: PubSub.PubSub<void>
+    pruning: Promise<void> | undefined
     spawning: Map<string, Promise<LSPClient.Info | undefined>>
+    subs: Map<string, { sub: FSWatcher; names: Set<string> }>
   }
 
   export interface Interface {
@@ -210,12 +217,25 @@ export namespace LSP {
             clients: [],
             servers,
             broken: new Set(),
+            pulse: yield* PubSub.unbounded<void>(),
+            pruning: undefined,
             spawning: new Map(),
+            subs: new Map(),
           }
 
+          yield* Stream.fromPubSub(s.pulse).pipe(
+            Stream.debounce("50 millis"),
+            Stream.runForEach(() => Effect.promise(() => scan(s))),
+            Effect.forkScoped,
+          )
+
           yield* Effect.addFinalizer(() =>
-            Effect.promise(async () => {
-              await Promise.all(s.clients.map((client) => client.shutdown()))
+            Effect.gen(function* () {
+              yield* PubSub.shutdown(s.pulse).pipe(Effect.ignore)
+              for (const item of s.subs.values()) {
+                item.sub.close()
+              }
+              yield* Effect.promise(() => Promise.all(s.clients.map((client) => client.shutdown())))
             }),
           )
 
@@ -225,6 +245,7 @@ export namespace LSP {
 
       const getClients = Effect.fnUntraced(function* (file: string) {
         if (!Instance.containsPath(file)) return [] as LSPClient.Info[]
+        yield* trim()
         const s = yield* InstanceState.get(state)
         return yield* Effect.promise(async () => {
           const extension = path.parse(file).ext || file
@@ -266,6 +287,7 @@ export namespace LSP {
             }
 
             s.clients.push(client)
+            sync(s)
             return client
           }
 
@@ -274,7 +296,8 @@ export namespace LSP {
 
             const root = await server.root(file)
             if (!root) continue
-            if (s.broken.has(root + server.id)) continue
+            const id = key(server.id, root)
+            if (s.broken.has(id)) continue
 
             const match = s.clients.find((x) => x.root === root && x.serverID === server.id)
             if (match) {
@@ -282,7 +305,7 @@ export namespace LSP {
               continue
             }
 
-            const inflight = s.spawning.get(root + server.id)
+            const inflight = s.spawning.get(id)
             if (inflight) {
               const client = await inflight
               if (!client) continue
@@ -290,12 +313,12 @@ export namespace LSP {
               continue
             }
 
-            const task = schedule(server, root, root + server.id)
-            s.spawning.set(root + server.id, task)
+            const task = schedule(server, root, id)
+            s.spawning.set(id, task)
 
             task.finally(() => {
-              if (s.spawning.get(root + server.id) === task) {
-                s.spawning.delete(root + server.id)
+              if (s.spawning.get(id) === task) {
+                s.spawning.delete(id)
               }
             })
 
@@ -315,7 +338,89 @@ export namespace LSP {
         return yield* Effect.promise(() => Promise.all(clients.map((x) => fn(x))))
       })
 
+      function sync(s: State) {
+        const next = new Map<string, Set<string>>()
+
+        for (const client of s.clients) {
+          const dir = path.dirname(client.root)
+          const names = next.get(dir) ?? new Set<string>()
+          names.add(path.basename(client.root))
+          next.set(dir, names)
+        }
+
+        for (const [dir, item] of s.subs) {
+          if (next.has(dir)) continue
+          s.subs.delete(dir)
+          item.sub.close()
+        }
+
+        for (const [dir, names] of next) {
+          const existing = s.subs.get(dir)
+          if (existing) {
+            existing.names = names
+            continue
+          }
+          try {
+            const sub = fswatch(
+              dir,
+              { persistent: false },
+              Instance.bind((_, file) => {
+                if (file) {
+                  const name = String(file)
+                  if (!s.subs.get(dir)?.names.has(name)) return
+                }
+                fire(s)
+              }),
+            )
+            sub.on(
+              "error",
+              Instance.bind(() => {
+                if (s.subs.get(dir)?.sub !== sub) return
+                s.subs.delete(dir)
+                sub.close()
+                fire(s)
+              }),
+            )
+            s.subs.set(dir, { sub, names })
+          } catch {}
+        }
+      }
+
+      function fire(s: State) {
+        Effect.runFork(PubSub.publish(s.pulse, undefined).pipe(Effect.ignore))
+      }
+
+      async function scan(s: State) {
+        if (s.pruning) return s.pruning
+
+        const task = (async () => {
+          const dead = (
+            await Promise.all(
+              s.clients.map(async (client) => ((await Filesystem.exists(client.root)) ? undefined : client)),
+            )
+          ).filter((client): client is LSPClient.Info => Boolean(client))
+          if (!dead.length) return
+
+          const ids = new Set(dead.map((client) => key(client.serverID, client.root)))
+          s.clients = s.clients.filter((client) => !ids.has(key(client.serverID, client.root)))
+          sync(s)
+          await Promise.all(dead.map((client) => client.shutdown().catch(() => undefined)))
+          await Bus.publish(Event.Updated, {})
+        })().finally(() => {
+          if (s.pruning === task) s.pruning = undefined
+        })
+
+        s.pruning = task
+        return task
+      }
+
+      const trim = Effect.fnUntraced(function* () {
+        const s = yield* InstanceState.get(state)
+        yield* Effect.promise(() => scan(s))
+      })
+
       const runAll = Effect.fnUntraced(function* <T>(fn: (client: LSPClient.Info) => Promise<T>) {
+        yield* trim()
         const s = yield* InstanceState.get(state)
         return yield* Effect.promise(() => Promise.all(s.clients.map((x) => fn(x))))
       })
@@ -325,6 +430,7 @@ export namespace LSP {
       })
 
       const status = Effect.fn("LSP.status")(function* () {
+        yield* trim()
         const s = yield* InstanceState.get(state)
         const result: Status[] = []
         for (const client of s.clients) {
@@ -346,7 +452,7 @@ export namespace LSP {
             if (server.extensions.length && !server.extensions.includes(extension)) continue
             const root = await server.root(file)
             if (!root) continue
-            if (s.broken.has(root + server.id)) continue
+            if (s.broken.has(key(server.id, root))) continue
             return true
           }
           return false

packages/opencode/test/fixture/lsp/fake-lsp-server.js

@@ -1,7 +1,36 @@
 // Simple JSON-RPC 2.0 LSP-like fake server over stdio
 // Implements a minimal LSP handshake and triggers a request upon notification
 
+const fs = require("fs")
 const net = require("net")
+const path = require("path")
+
+const mark = process.argv[2]
+
+if (mark) {
+  try {
+    process.chdir(path.dirname(mark))
+  } catch {}
+}
+
+function writeMark() {
+  if (!mark) return
+  try {
+    fs.writeFileSync(mark, "exit")
+  } catch {}
+}
+
+process.on("exit", writeMark)
+process.on("SIGTERM", () => {
+  writeMark()
+  process.exit(0)
+})
+process.on("SIGINT", () => {
+  writeMark()
+  process.exit(0)
+})
+
+setInterval(() => {}, 1000)
 
 let nextId = 1
 

packages/opencode/test/lsp/cleanup-effect.test.ts

@@ -0,0 +1,68 @@
+import { afterEach, describe, expect } from "bun:test"
+import { Deferred, Effect, Layer } from "effect"
+import { Bus } from "../../src/bus"
+import path from "path"
+import { setTimeout as sleep } from "node:timers/promises"
+import * as CrossSpawnSpawner from "../../src/effect/cross-spawn-spawner"
+import { AppFileSystem } from "../../src/filesystem"
+import { LSP } from "../../src/lsp"
+import { Instance } from "../../src/project/instance"
+import { provideTmpdirInstance } from "../fixture/fixture"
+import { testEffect } from "../lib/effect"
+
+afterEach(async () => {
+  await Instance.disposeAll()
+})
+
+const it = testEffect(Layer.mergeAll(LSP.defaultLayer, CrossSpawnSpawner.defaultLayer, AppFileSystem.defaultLayer))
+const server = path.join(import.meta.dir, "../fixture/lsp/fake-lsp-server.js")
+
+describe("LSP cleanup", () => {
+  it.live("shuts down clients when their root is deleted", () =>
+    provideTmpdirInstance((dir) =>
+      Effect.gen(function* () {
+        const fs = yield* AppFileSystem.Service
+        const mark = path.join(path.dirname(dir), `${path.basename(dir)}.exit`)
+        const file = path.join(dir, "test.ts")
+
+        yield* Effect.addFinalizer(() => fs.remove(mark, { force: true }).pipe(Effect.ignore))
+        yield* fs.writeWithDirs(
+          path.join(dir, "opencode.json"),
+          JSON.stringify({
+            $schema: "https://opencode.ai/config.json",
+            lsp: {
+              typescript: { disabled: true },
+              fake: {
+                command: [process.execPath, server, mark],
+                extensions: [".ts"],
+              },
+            },
+          }),
+        )
+        yield* fs.writeWithDirs(file, "export {}\n")
+        yield* LSP.Service.use((svc) => svc.touchFile(file))
+        expect(yield* LSP.Service.use((svc) => svc.status())).toHaveLength(1)
+
+        const done = yield* Deferred.make<void>()
+        const off = Bus.subscribe(LSP.Event.Updated, () => {
+          Deferred.doneUnsafe(done, Effect.void)
+        })
+        yield* Effect.addFinalizer(() => Effect.sync(off))
+
+        yield* fs.remove(dir, { recursive: true, force: true })
+        yield* Deferred.await(done).pipe(Effect.timeout("2 seconds"))
+
+        const stopped = yield* Effect.promise(async () => {
+          for (const _ of Array.from({ length: 20 })) {
+            if (await fs.exists(mark)) return true
+            await sleep(50)
+          }
+          return false
+        })
+
+        expect(stopped).toBe(true)
+        expect(yield* LSP.Service.use((svc) => svc.status())).toHaveLength(0)
+      }),
+    ),
+  )
+})