omp-debug/lib/dap.js

// 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)
  }
}