1/**

2 * Generic stdio language-server backend for `ctx.lsp`. One plugin instance configures a named table

3 * of server commands and registers one isolated provider for each entry. Every provider lazily

4 * single-flights one server process per canonical workspace target, serves transient-open queries

5 * through it, and replaces a selected transport that fails before or during the next read-only

6 * query. Providers read sources through `ctx.fs` and launch servers through

7 * `ctx.subprocess`, so both local and remote implementations share one host.

8 *

9 * Namespace plugin (named exports, no default export). Lifecycle is effect-scoped: disposal

10 * unregisters from `ctx.lsp` and tears down every live server.

11 * @module @deepseek-ai/dsh-lsp-stdio

12 */

13

14import type { Context } from '@deepseek-ai/cordis'

15import z from '@deepseek-ai/schemastery'

16import { LspError, LspProviderId } from '@deepseek-ai/dsh-lsp'

17import type {

18 LspProvider,

19 LspProviderQuery,

20 LspQueryResult,

21} from '@deepseek-ai/dsh-lsp'

22import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'

23import { abortable, abortError } from './abort.ts'

24import { canonicalizeWorkspace, readHostSource } from './host.ts'

25import type { HostWorkspace } from './host.ts'

26import { LspInstance } from './instance.ts'

27import type { ConnectionSpawner } from './connection.ts'

28import type { InstanceSpec } from './instance.ts'

29

30export { canonicalizeWorkspace, readHostSource } from './host.ts'

31export { encodeMessage, MessageDecoder } from './framing.ts'

32export {

33 negotiatePositionEncoding,

34 normalizeHover,

35 normalizeLocations,

36 requestMethod,

37 supportsOperation,

38 supportsTransientOpen,

39} from './translate.ts'

40export { LspInstance } from './instance.ts'

41export { LspConnection } from './connection.ts'

42

43/** Cordis plugin name for loader diagnostics. */

44export const name = 'lsp-stdio'

45

46/** Services required by this plugin. */

47export const inject = ['fs', 'lsp', 'subprocess']

48

49const DEFAULT_MAX_MESSAGE_BYTES = 16_000_000

50const DEFAULT_MAX_STDERR_BYTES = 1_000_000

51const DEFAULT_MAX_DOCUMENT_BYTES = 4_000_000

52const DEFAULT_SHUTDOWN_TIMEOUT_MS = 5_000

53const DEFAULT_KILL_GRACE_MS = 2_000

54

55/** One configured local language server and its host bounds. */

56export interface LspLocalServerConfig {

57 /** Executable to spawn (absolute, or resolved on PATH at load). */

58 command: string

59 /** Lowercase leading-dot extension → LSP language id (e.g. `{ '.ts': 'typescript' }`). */

60 extensionToLanguage: Record<string, string>

61 /** Arguments passed to the executable (no shell). Default `[]`. */

62 args?: string[]

63 /** Extra env vars merged on top of the scrubbed ambient env. Default `{}`. */

64 env?: Record<string, string>

65 /** Static `initialize` options forwarded to the server. Default `null`. */

66 initializationOptions?: unknown

67 /** Static answer to every `workspace/configuration` item. Default `null`. */

68 configuration?: unknown

69 /** Largest single framed message accepted from the server (bytes). Default 16000000. */

70 maxMessageBytes?: number

71 /** Largest stderr tail retained for diagnostics (bytes). Default 1000000. */

72 maxStderrBytes?: number

73 /** Largest source file this host will open (bytes). Default 4000000. */

74 maxDocumentBytes?: number

75 /** Graceful `shutdown`/`exit` budget before escalation (ms). Default 5000. */

76 shutdownTimeoutMs?: number

77 /** Request-cancel and SIGTERM→SIGKILL grace (ms). Default 2000. */

78 killGraceMs?: number

79}

80

81/** Plugin configuration: provider id → local language-server configuration. */

82export interface Config {

83 /** Non-empty table of stable provider ids to independent local server configurations. */

84 servers: Record<string, LspLocalServerConfig>

85}

86

87/** One server config after schemastery fills every default. */

88type ResolvedServerConfig = Required<LspLocalServerConfig>

89type WorkspaceKey = HostWorkspace['target']['targetKey']

90

91const LspLocalServerConfig: z<LspLocalServerConfig> = z.object({

92 command: z.string().required(),

93 args: z.array(String).default([]),

94 env: z.dict(String).default({}),

95 extensionToLanguage: z.dict(String).required(),

96 initializationOptions: z.any().default(null),

97 configuration: z.any().default(null),

98 maxMessageBytes: z.number().default(DEFAULT_MAX_MESSAGE_BYTES),

99 maxStderrBytes: z.number().default(DEFAULT_MAX_STDERR_BYTES),

100 maxDocumentBytes: z.number().default(DEFAULT_MAX_DOCUMENT_BYTES),

101 shutdownTimeoutMs: z.number().max(MAX_TIMER_DELAY_MS).default(DEFAULT_SHUTDOWN_TIMEOUT_MS),

102 killGraceMs: z.number().max(MAX_TIMER_DELAY_MS).default(DEFAULT_KILL_GRACE_MS),

103})

104

105export const Config: z<Config> = z.object({

106 servers: z.dict(LspLocalServerConfig).required(),

107})

108

109/** Propagate teardown failures only after every sibling has settled. */

110function throwTeardownFailures(results: readonly PromiseSettledResult<void>[], message: string): void {

111 const failures: unknown[] = []

112 for (const result of results) {

113 if (result.status === 'rejected') failures.push(result.reason)

114 }

115 if (failures.length === 1) throw failures[0]

116 if (failures.length > 1) throw new AggregateError(failures, message)

117}

118

119/**

120 * Register the configured stdio LSP providers. Resolves every executable at load (after credential

121 * scrubbing) before publishing any provider; each process launches lazily on its first matching

122 * query.

123 * @param ctx - the plugin context carrying `fs`, `lsp`, and `subprocess`.

124 * @param config - the resolved plugin configuration (schemastery has filled every default).

125 */

126export async function apply(ctx: Context, config: Config): Promise<void> {

127 const entries = Object.entries(config.servers)

128 if (entries.length === 0) throw new Error('lsp-stdio: servers must contain at least one server')

129

130 const setupAbort = new AbortController()

131 const stopSetupCancellation = ctx.on('internal/plugin', (fiber) => {

132 // An async plugin callback must observe its own disposal before Cordis can

133 // run effect cleanup, because unload otherwise waits for this callback.

134 if (fiber === ctx.fiber && fiber.uid === null) {

135 setupAbort.abort(new Error('lsp-stdio setup disposed'))

136 }

137 })

138

139 // Resolve every server-local setting before registration so a bad later command or bound cannot

140 // publish an earlier provider. Registry-level mapping conflicts are rolled back below.

141 const providers = await (async () => {

142 const lookups = entries.map(async ([providerId, rawConfig]) => {

143 if (providerId.trim() === '') throw new Error('lsp-stdio: server ids must be non-empty strings')

144 const resolved = rawConfig as ResolvedServerConfig

145 validateServerConfig(providerId, resolved)

146 const executable = await ctx.subprocess.resolveExecutable(

147 resolved.command,

148 resolved.env,

149 setupAbort.signal,

150 )

151 setupAbort.signal.throwIfAborted()

152 return new LocalLspProvider(

153 providerId,

154 ctx.fs,

155 resolved,

156 executable,

157 spec => ctx.subprocess.spawn(spec),

158 )

159 })

160 try {

161 return await Promise.all(lookups)

162 } catch (error: unknown) {

163 setupAbort.abort(error)

164 await Promise.allSettled(lookups)

165 throw error

166 } finally {

167 stopSetupCancellation()

168 }

169 })()

170

171 ctx.effect(() => {

172 const disposers: Array<() => void> = []

173 try {

174 for (const provider of providers) disposers.push(ctx.lsp.registerProvider(provider))

175 } catch (error) {

176 for (const dispose of disposers.reverse()) dispose()

177 throw error

178 }

179 return async () => {

180 // Remove every route before process teardown so no new query can enter a draining provider.

181 for (const dispose of disposers.reverse()) dispose()

182 const results = await Promise.allSettled(providers.map(provider => provider.disposeAll()))

183 throwTeardownFailures(results, 'lsp-stdio provider teardown failed')

184 }

185 }, 'lsp-stdio.registerProviders')

186}

187

188/** Validate one resolved server entry before any provider in the table is registered. */

189function validateServerConfig(providerId: string, resolved: ResolvedServerConfig): void {

190 // Teardown budgets feed `deadline()`, whose `<= 0` is the internal no-timeout sentinel; a

191 // nonpositive value would let a server that ignores shutdown hang disposal forever. Fail at load.

192 assertTimer(providerId, 'shutdownTimeoutMs', resolved.shutdownTimeoutMs)

193 assertTimer(providerId, 'killGraceMs', resolved.killGraceMs)

194 // Byte caps must be positive: a nonpositive stderr cap defeats the retained-tail bound

195 // (`slice(-0)` keeps everything), `maxMessageBytes: 0` makes every response fatal, and a bad

196 // document cap fails later in the read path instead of at load.

197 assertPositiveInteger(providerId, 'maxStderrBytes', resolved.maxStderrBytes)

198 assertPositiveInteger(providerId, 'maxMessageBytes', resolved.maxMessageBytes)

199 assertPositiveInteger(providerId, 'maxDocumentBytes', resolved.maxDocumentBytes)

200}

201

202/** Reject a timer value Node would clamp instead of scheduling as configured. */

203function assertTimer(providerId: string, name: string, value: number): void {

204 if (!Number.isInteger(value) || value < 1 || value > MAX_TIMER_DELAY_MS) {

205 throw new Error(`lsp-stdio: servers.${providerId}.${name} must be a positive integer no greater than ${MAX_TIMER_DELAY_MS}`)

206 }

207}

208

209/** Reject a nonpositive or non-integer config value at load, so misconfiguration fails loud. */

210function assertPositiveInteger(providerId: string, name: string, value: number): void {

211 if (!Number.isInteger(value) || value < 1) {

212 throw new Error(`lsp-stdio: servers.${providerId}.${name} must be a positive integer`)

213 }

214}

215

216/** A pooled generic provider: one server process per canonical workspace, created on demand. */

217class LocalLspProvider implements LspProvider {

218 readonly id: LspProviderId

219 readonly extensionToLanguage: Readonly<Record<string, string>>

220 /** One live instance per stable canonical workspace identity. */

221 private readonly instances = new Map<WorkspaceKey, LspInstance>()

222 /** One complete source-read→open→query→close serialization tail per canonical workspace. */

223 private readonly queues = new Map<WorkspaceKey, Promise<void>>()

224 /** Workspace canonicalizations that have not entered a provider-owned queue yet. */

225 private readonly workspaceLookups = new Set<Promise<void>>()

226 private readonly lifetime = new AbortController()

227 private disposed = false

228

229 constructor(

230 providerId: string,

231 private readonly fs: Context['fs'],

232 private readonly config: ResolvedServerConfig,

233 private readonly executable: string,

234 private readonly spawner: ConnectionSpawner,

235 ) {

236 this.id = LspProviderId(providerId)

237 this.extensionToLanguage = config.extensionToLanguage

238 }

239

240 /** Read the disposed flag through a method so a `query()` await cannot narrow it to a literal. */

241 private isDisposed(): boolean {

242 return this.disposed

243 }

244

245 /** Reject work that cannot publish or use a provider-owned instance. */

246 private assertActive(signal?: AbortSignal): void {

247 /* v8 ignore next -- the seam unregisters this provider before disposal; direct in-flight calls

248 exercise the post-await check instead. */

249 if (this.isDisposed()) throw new LspError('lsp-stdio provider is disposed', 'LSP_DISPOSED')

250 if (signal?.aborted) throw abortError(signal)

251 }

252

253 /** Fuse caller cancellation with provider disposal for every filesystem and protocol await. */

254 private querySignal(signal?: AbortSignal): AbortSignal {

255 return signal === undefined

256 ? this.lifetime.signal

257 : AbortSignal.any([signal, this.lifetime.signal])

258 }

259

260 async query(request: LspProviderQuery, signal?: AbortSignal): Promise<LspQueryResult> {

261 // Honor an already-aborted signal before provider I/O so a canceled request never starts a server.

262 this.assertActive(signal)

263 const querySignal = this.querySignal(signal)

264 const workspaceResult = canonicalizeWorkspace(this.fs, request.workspaceRoot, querySignal)

265 const workspaceLookup = workspaceResult.then(() => undefined, () => undefined)

266 this.workspaceLookups.add(workspaceLookup)

267 let workspace: HostWorkspace

268 try {

269 workspace = await workspaceResult

270 } finally {

271 this.workspaceLookups.delete(workspaceLookup)

272 }

273 this.assertActive(querySignal)

274 const workspaceKey = workspace.target.targetKey

275 return this.enqueue(workspaceKey, querySignal, async () => {

276 this.assertActive(querySignal)

277 // Read inside the workspace queue but before spawning: a queued query sees current bytes when

278 // its turn starts, while an invalid source still cannot leave an idle process pooled.

279 const source = await readHostSource(this.fs, request.filePath, workspace, this.config.maxDocumentBytes, querySignal)

280 // Disposal may have snapshotted the instance map while host I/O was pending. Re-check before a

281 // synchronous get-or-create so every spawned process remains owned by teardown.

282 this.assertActive(querySignal)

283 let instance = this.instanceFor(workspaceKey, workspace)

284 let canRetryTransport = true

285 for (;;) {

286 const [queryOutcome] = await Promise.allSettled([

287 instance.query(request, source, querySignal),

288 ])

289 let teardownOutcome: PromiseSettledResult<void> | undefined

290 if (instance.dead) {

291 ;[teardownOutcome] = await Promise.allSettled([instance.dispose()])

292 // A dead instance is never reusable, even when its final quiescence observation fails.

293 this.evictIfCurrent(workspaceKey, instance)

294 }

295 if (teardownOutcome?.status === 'rejected') {

296 if (queryOutcome.status === 'rejected') {

297 throw new AggregateError(

298 [queryOutcome.reason, teardownOutcome.reason],

299 'LSP operation and teardown failed',

300 )

301 }

302 throw teardownOutcome.reason

303 }

304 if (queryOutcome.status === 'fulfilled') return queryOutcome.value

305 // A selected child can have died while idle or fail during the next write. Queries are

306 // read-only, so replace that transport once and retry transparently after clean disposal.

307 if (!canRetryTransport || !instance.isTransportFailure(queryOutcome.reason)) {

308 throw queryOutcome.reason

309 }

310 canRetryTransport = false

311 this.assertActive(querySignal)

312 instance = this.instanceFor(workspaceKey, workspace)

313 }

314 })

315 }

316

317 /** Serialize one complete query lifecycle for a canonical workspace. */

318 private enqueue<T>(workspace: WorkspaceKey, signal: AbortSignal | undefined, run: () => Promise<T>): Promise<T> {

319 const previous = this.queues.get(workspace) ?? Promise.resolve()

320 const result = abortable(previous, signal).then(run)

321 // The tail follows the actual prior work even when this caller aborts its wait. It never rejects,

322 // so later callers serialize without inheriting an earlier query's outcome.

323 const tail = previous.then(() => result).then(() => undefined, () => undefined)

324 this.queues.set(workspace, tail)

325 void tail.then(() => {

326 if (this.queues.get(workspace) === tail) this.queues.delete(workspace)

327 })

328 return result

329 }

330

331 /** Return or synchronously publish the one instance for a canonical workspace. */

332 private instanceFor(workspaceKey: WorkspaceKey, workspace: HostWorkspace): LspInstance {

333 this.assertActive()

334 const existing = this.instances.get(workspaceKey)

335 if (existing !== undefined) return existing

336 const created = this.createInstance(workspace)

337 this.instances.set(workspaceKey, created)

338 return created

339 }

340

341 /** Drop the slot iff it still contains this instance. */

342 private evictIfCurrent(workspace: WorkspaceKey, instance: LspInstance): void {

343 /* v8 ignore next -- mismatch requires another query to replace the slot before this finally runs. */

344 if (this.instances.get(workspace) === instance) this.instances.delete(workspace)

345 }

346

347 private createInstance(workspace: HostWorkspace): LspInstance {

348 const spec: InstanceSpec = {

349 command: this.executable,

350 args: this.config.args,

351 cwd: workspace.canonicalPath,

352 workspaceUri: workspace.fileUrl,

353 env: this.config.env,

354 configuration: this.config.configuration,

355 initializationOptions: this.config.initializationOptions,

356 maxMessageBytes: this.config.maxMessageBytes,

357 maxStderrBytes: this.config.maxStderrBytes,

358 shutdownTimeoutMs: this.config.shutdownTimeoutMs,

359 killGraceMs: this.config.killGraceMs,

360 }

361 return new LspInstance(spec, this.spawner)

362 }

363

364 /** Dispose every live instance and block further queries. */

365 async disposeAll(): Promise<void> {

366 this.disposed = true

367 this.lifetime.abort(new LspError('lsp-stdio provider is disposed', 'LSP_DISPOSED'))

368 const live = [...this.instances.values()]

369 const draining = [...this.queues.values()]

370 const resolving = [...this.workspaceLookups]

371 this.instances.clear()

372 const results = await Promise.allSettled([

373 ...live.map(instance => instance.dispose()),

374 ...draining,

375 ...resolving,

376 ])

377 this.queues.clear()

378 this.workspaceLookups.clear()

379 throwTeardownFailures(results, 'lsp-stdio instance teardown failed')

380 }

381}