// Debug Adapter Protocol client (stdio or TCP transport, Content-Length framing). import { spawn } from 'node:child_process' import { EventEmitter } from 'node:events' import net from 'node:net' export class DapClient extends EventEmitter { constructor(name) { super() this.name = name this.seq = 0 this.pending = new Map() this.buf = Buffer.alloc(0) this.output = [] this.outputBytes = 0 this.closed = false this.capabilities = {} this.reverse = {} // command -> async handler(args) -> body } /** stdio adapter */ static spawnStdio(name, command, args, { cwd, env } = {}) { const c = new DapClient(name) c.proc = spawn(command, args, { cwd, env: { ...process.env, ...env }, stdio: ['pipe', 'pipe', 'pipe'], detached: true }) c.proc.stdout.on('data', (d) => c.onData(d)) c.proc.stderr.on('data', (d) => c.log('adapter', String(d))) c.proc.on('exit', (code, sig) => c.close(`adapter exited (${sig ?? code})`)) c.proc.on('error', (e) => c.close(e.message)) c.write = (s) => { if (c.proc.stdin.writable) c.proc.stdin.write(s) } return c } /** TCP adapter: optionally spawn a server process, then connect (with retries). */ static async connectTcp(name, { host = '127.0.0.1', port, spawnCmd, spawnArgs, cwd, env, timeoutMs = 15000 }) { const c = new DapClient(name) if (spawnCmd) { c.proc = spawn(spawnCmd, spawnArgs, { cwd, env: { ...process.env, ...env }, stdio: ['ignore', 'pipe', 'pipe'], detached: true }) c.proc.stdout.on('data', (d) => c.log('adapter', String(d))) c.proc.stderr.on('data', (d) => c.log('adapter', String(d))) c.proc.on('exit', (code, sig) => c.close(`adapter exited (${sig ?? code})`)) } const deadline = Date.now() + timeoutMs for (;;) { try { c.sock = await new Promise((resolve, reject) => { const s = net.connect({ host, port }, () => resolve(s)) s.once('error', reject) }) break } catch (e) { if (Date.now() > deadline || c.closed) { c.close(`connect ${host}:${port} failed: ${e.message}`); throw new Error(`${name}: could not connect to ${host}:${port} (${e.message})`) } await new Promise((r) => setTimeout(r, 150)) } } c.sock.on('data', (d) => c.onData(d)) c.sock.on('close', () => c.close('socket closed')) c.sock.on('error', () => {}) c.write = (s) => { if (!c.sock.destroyed) c.sock.write(s) } return c } log(category, text) { if (!text) return this.output.push({ category, text, at: Date.now() }) this.outputBytes += text.length while (this.outputBytes > 200_000 && this.output.length > 1) this.outputBytes -= this.output.shift().text.length } onData(chunk) { this.buf = Buffer.concat([this.buf, chunk]) for (;;) { const sep = this.buf.indexOf('\r\n\r\n') if (sep < 0) return const m = /Content-Length:\s*(\d+)/i.exec(this.buf.subarray(0, sep).toString('ascii')) if (!m) { this.buf = this.buf.subarray(sep + 4); continue } const len = Number(m[1]) if (this.buf.length < sep + 4 + len) return const body = this.buf.subarray(sep + 4, sep + 4 + len).toString('utf8') this.buf = this.buf.subarray(sep + 4 + len) let msg try { msg = JSON.parse(body) } catch { continue } this.onMessage(msg) } } async onMessage(msg) { if (msg.type === 'response') { const p = this.pending.get(msg.request_seq) if (!p) return this.pending.delete(msg.request_seq) clearTimeout(p.timer) if (msg.success) p.resolve(msg.body ?? {}) else p.reject(Object.assign(new Error(msg.body?.error?.format ?? msg.message ?? `${p.command} failed`), { body: msg.body })) return } if (msg.type === 'event') { if (msg.event === 'output') this.log(msg.body?.category ?? 'console', msg.body?.output ?? '') this.emit('event', msg) this.emit(msg.event, msg.body ?? {}) return } if (msg.type === 'request') { const h = this.reverse[msg.command] try { const body = h ? await h(msg.arguments ?? {}) : {} this.send({ type: 'response', request_seq: msg.seq, success: !!h, command: msg.command, body, ...(h ? {} : { message: 'not supported' }) }) } catch (e) { this.send({ type: 'response', request_seq: msg.seq, success: false, command: msg.command, message: e.message }) } } } send(obj) { const s = JSON.stringify({ seq: ++this.seq, ...obj }) this.write?.(`Content-Length: ${Buffer.byteLength(s, 'utf8')}\r\n\r\n${s}`) return this.seq } request(command, args, timeoutMs = 30000) { if (this.closed) return Promise.reject(new Error(`${this.name}: session closed (${this.closeReason})`)) return new Promise((resolve, reject) => { const seq = this.send({ type: 'request', command, arguments: args }) const timer = setTimeout(() => { this.pending.delete(seq); reject(new Error(`${this.name}: ${command} timed out after ${timeoutMs}ms`)) }, timeoutMs) this.pending.set(seq, { resolve, reject, timer, command }) }) } /** Wait for an event, optionally filtered; resolves undefined on timeout. */ waitFor(event, { timeoutMs = 10000, filter = () => true, signal } = {}) { return new Promise((resolve) => { const on = (body) => { if (filter(body)) { done(); resolve(body) } } const onClose = () => { done(); resolve(undefined) } const timer = setTimeout(() => { done(); resolve(undefined) }, timeoutMs) const onAbort = () => { done(); resolve(undefined) } const done = () => { clearTimeout(timer); this.off(event, on); this.off('closed', onClose); signal?.removeEventListener?.('abort', onAbort) } this.on(event, on) this.once('closed', onClose) signal?.addEventListener?.('abort', onAbort, { once: true }) }) } close(reason = 'closed') { if (this.closed) return this.closed = true this.closeReason = reason for (const p of this.pending.values()) { clearTimeout(p.timer); p.reject(new Error(`${this.name}: ${reason}`)) } this.pending.clear() try { this.sock?.destroy() } catch {} // Kill the adapter (and anything it spawned) — a lingering adapter keeps the host's event loop busy. const p = this.proc if (p && p.exitCode === null && p.signalCode === null) { try { process.kill(-p.pid, 'SIGTERM') } catch { try { p.kill('SIGTERM') } catch {} } const t = setTimeout(() => { try { process.kill(-p.pid, 'SIGKILL') } catch { try { p.kill('SIGKILL') } catch {} } }, 1500) t.unref?.() } for (const s of [p?.stdin, p?.stdout, p?.stderr]) { try { s?.destroy() } catch {} } this.emit('closed', reason) } }