bare-operating-system/packages/bare-os-booter/lib/bare-os-ipc.js

/**
 * In-memory named FIFOs under logical `/run/bare-os/ipc/<name>`.
 * One write wakes one blocking read; queued writes are delivered in order.
 * Optional JSON-RPC-style envelopes: objects with `bareOsRpc`, `id`, and `method`/`result`/`error`.
 * Optional fan-out channels: multiple subscribers each receive a copy (bounded per subscriber).
 */

import b4a from 'b4a'

class FifoChannel {
  constructor() {
    /** @type {Uint8Array[]} */
    this.queue = []
    /** @type {{ resolve: (v: Uint8Array) => void, cleanup?: () => void }[]} */
    this.waiters = []
    /** @type {number} */
    this.queuedBytes = 0
  }

  /**
   * @param {Uint8Array | ArrayBuffer} buf
   * @param {number} maxQueuedBytes
   */
  push(buf, maxQueuedBytes) {
    const u8 = buf instanceof Uint8Array ? buf : new Uint8Array(buf)
    if (u8.byteLength > maxQueuedBytes) {
      throw new Error('fifo payload exceeds BARE_OS_IPC_MAX_BYTES')
    }
    if (this.waiters.length > 0) {
      const w = this.waiters.shift()
      try {
        w.cleanup?.()
      } catch {
        /* ignore */
      }
      w.resolve(u8)
      return
    }
    const nextQueued = this.queuedBytes + u8.byteLength
    if (nextQueued > maxQueuedBytes) {
      throw new Error('fifo backlog exceeds BARE_OS_IPC_MAX_BYTES')
    }
    this.queue.push(u8)
    this.queuedBytes = nextQueued
  }

  /**
   * @param {AbortSignal} [signal]
   * @returns {Promise<Uint8Array>}
   */
  take(signal) {
    if (this.queue.length > 0) {
      const u8 = this.queue.shift()
      this.queuedBytes -= u8.byteLength
      return Promise.resolve(u8)
    }
    return new Promise((resolve, reject) => {
      /** @type {{ resolve: typeof resolve, cleanup?: () => void }} */
      const entry = { resolve }
      if (signal) {
        if (signal.aborted) {
          reject(
            Object.assign(new Error('Aborted'), { name: 'AbortError' })
          )
          return
        }
        const onAbort = () => {
          const i = this.waiters.indexOf(entry)
          if (i >= 0) this.waiters.splice(i, 1)
          reject(
            Object.assign(new Error('Aborted'), { name: 'AbortError' })
          )
        }
        signal.addEventListener('abort', onAbort, { once: true })
        entry.cleanup = () => {
          try {
            signal.removeEventListener('abort', onAbort)
          } catch {
            /* ignore */
          }
        }
      }
      this.waiters.push(entry)
    })
  }
}

class FanoutHub {
  /**
   * @param {number} maxPerSubBytes
   */
  constructor(maxPerSubBytes) {
    this.maxPerSubBytes = maxPerSubBytes
    /** @type {FifoChannel[]} */
    this.subscribers = []
  }

  /**
   * @param {Uint8Array} u8
   */
  publish(u8) {
    let drops = 0
    for (const fifo of this.subscribers) {
      try {
        fifo.push(u8, this.maxPerSubBytes)
      } catch {
        /* drop for this subscriber if backlog full */
        drops++
      }
    }
    return drops
  }

  subscribe() {
    const fifo = new FifoChannel()
    this.subscribers.push(fifo)
    return {
      take: (sig) => fifo.take(sig),
      dispose: () => {
        const i = this.subscribers.indexOf(fifo)
        if (i >= 0) this.subscribers.splice(i, 1)
      }
    }
  }
}

/**
 * @param {string} name
 */
function assertSafeIpcName(name) {
  if (!name || name.length > 128) {
    throw new Error('invalid fifo name')
  }
  if (!/^[a-zA-Z0-9._-]+$/.test(name)) {
    throw new Error('fifo name must match [a-zA-Z0-9._-]+')
  }
}

const DEFAULT_JSON_RPC_MAX = 256 * 1024

/**
 * @param {{ maxFifoBytes?: number, maxChannels?: number, perChannelMaxBytes?: Map<string, number>, ipcRpcToken?: string | null, enableFanout?: boolean, maxJsonRpcLineBytes?: number, maxWaitersPerChannel?: number, posixMqDefaultMaxmsg?: number, posixMqDefaultMaxBytes?: number, posixMqMaxmsgCeiling?: number }} [opts]
 */
export function createBareOsIpc(opts = {}) {
  const maxFifoBytes =
    typeof opts.maxFifoBytes === 'number' && opts.maxFifoBytes > 0
      ? Math.min(opts.maxFifoBytes, 16 * 1024 * 1024)
      : 1024 * 1024

  const maxChannels =
    typeof opts.maxChannels === 'number' && opts.maxChannels > 0
      ? Math.min(65536, Math.floor(opts.maxChannels))
      : 4096

  const perChannelMax =
    opts.perChannelMaxBytes instanceof Map ? opts.perChannelMaxBytes : null

  const maxWaitersPerChannel =
    typeof opts.maxWaitersPerChannel === 'number' && opts.maxWaitersPerChannel > 0
      ? Math.min(65536, Math.floor(opts.maxWaitersPerChannel))
      : 1024

  /** @type {{ fifoCreates: number, fifoCreateDeniedQuota: number, fifoPushDenied: number, fifoTakeAborted: number, fifoTakeDeniedWaiters: number, fanoutDrop: number, mqFull: number, backpressurePause: number, backpressureResume: number, rpcRequests: number, rpcTimeouts: number, rpcResponses: number, rpcLateResponses: number, jsonBatchSends: number }} */
  const telemetry = {
    fifoCreates: 0,
    fifoCreateDeniedQuota: 0,
    fifoPushDenied: 0,
    fifoTakeAborted: 0,
    fifoTakeDeniedWaiters: 0,
    fanoutDrop: 0,
    mqFull: 0,
    backpressurePause: 0,
    backpressureResume: 0,
    rpcRequests: 0,
    rpcTimeouts: 0,
    rpcResponses: 0,
    rpcLateResponses: 0,
    jsonBatchSends: 0
  }

  const limitFor = (name) => {
    if (perChannelMax && perChannelMax.has(name)) {
      const n = perChannelMax.get(name)
      if (typeof n === 'number' && n > 0)
        return Math.min(n, 16 * 1024 * 1024)
    }
    return maxFifoBytes
  }

  const maxJsonLine =
    typeof opts.maxJsonRpcLineBytes === 'number' && opts.maxJsonRpcLineBytes > 0
      ? Math.min(opts.maxJsonRpcLineBytes, 1024 * 1024)
      : DEFAULT_JSON_RPC_MAX
  const backpressureHighWaterBytes = Math.max(
    1024,
    Math.floor(maxFifoBytes * 0.8)
  )
  const backpressureLowWaterBytes = Math.max(
    512,
    Math.floor(maxFifoBytes * 0.5)
  )

  const rpcToken =
    opts.ipcRpcToken != null && String(opts.ipcRpcToken).length > 0
      ? String(opts.ipcRpcToken)
      : null

  const fanoutOn = opts.enableFanout !== false

  /** @type {Map<string, FifoChannel>} */
  const channels = new Map()
  /** @type {Map<string, string>} */
  const channelScopes = new Map()
  /** @type {Map<string, FanoutHub>} */
  const fanouts = new Map()
  /** @type {Set<string>} */
  const pausedChannels = new Set()
  /** @type {Map<string, { ready: boolean, waiters: Array<(v: true) => void> }>} */
  const channelReady = new Map()
  /** @type {Map<string, { resolve: (v: Record<string, unknown>) => void, reject: (e: Error) => void, to: ReturnType<typeof setTimeout> }>} */
  const pendingRequests = new Map()
  /** @type {Map<string, { failures: number, total: number, openUntilMs: number }>} */
  const rpcBreakerByMethod = new Map()
  let nextRequestId = 1

  /** @type {Map<string, { maxmsg: number, maxBytes: number, nextSeq: number, msgs: { prio: number, seq: number, data: Uint8Array }[], curBytes: number }>} */
  const mqQueues = new Map()

  const mqMsgCeiling =
    typeof opts.posixMqMaxmsgCeiling === 'number' &&
    opts.posixMqMaxmsgCeiling >= 32
      ? Math.min(65536, Math.floor(opts.posixMqMaxmsgCeiling))
      : 8192
  const mqDefaultMaxmsg =
    typeof opts.posixMqDefaultMaxmsg === 'number' &&
    opts.posixMqDefaultMaxmsg >= 1
      ? Math.min(mqMsgCeiling, Math.floor(opts.posixMqDefaultMaxmsg))
      : 32
  const mqDefaultMaxBytes =
    typeof opts.posixMqDefaultMaxBytes === 'number' &&
    opts.posixMqDefaultMaxBytes >= 256
      ? Math.min(1024 * 1024, Math.floor(opts.posixMqDefaultMaxBytes))
      : 65536

  return {
    /**
     * @param {string} name
     * @param {{ scope?: string }} [opts]
     */
    create(name, opts) {
      assertSafeIpcName(name)
      if (!channels.has(name)) {
        if (channels.size >= maxChannels) {
          telemetry.fifoCreateDeniedQuota++
          throw new Error('bare-os ipc: channel quota exceeded')
        }
        channels.set(name, new FifoChannel())
        telemetry.fifoCreates++
      }
      if (opts && opts.scope != null && String(opts.scope).trim()) {
        channelScopes.set(name, String(opts.scope).trim().slice(0, 64))
      }
      const row = channelReady.get(name)
      if (!row) {
        channelReady.set(name, { ready: true, waiters: [] })
      } else {
        row.ready = true
        for (const w of row.waiters.splice(0, row.waiters.length)) w(true)
      }
    },

    /**
     * @param {string} name
     */
    has(name) {
      return channels.has(name)
    },

    list() {
      return [...channels.keys()].sort()
    },

    /**
     * @param {string} name
     */
    remove(name) {
      channels.delete(name)
      channelScopes.delete(name)
      pausedChannels.delete(name)
      channelReady.delete(name)
    },

    /**
     * Assign a synthetic process-group id to a FIFO (POSIX setpgid analog for IPC routing).
     * @param {string} channelName
     * @param {number} pgid
     */
    assignProcessGroup(channelName, pgid) {
      assertSafeIpcName(channelName)
      const n = Number(pgid)
      if (!Number.isFinite(n) || n < 0 || n > 0xffffffff) {
        throw new Error('invalid pgid')
      }
      if (!channels.has(channelName)) {
        if (channels.size >= maxChannels) {
          telemetry.fifoCreateDeniedQuota++
          throw new Error('bare-os ipc: channel quota exceeded')
        }
        channels.set(channelName, new FifoChannel())
        telemetry.fifoCreates++
      }
      channelScopes.set(channelName, 'pgid:' + String(Math.floor(n)))
    },

    /**
     * Push a virtual signal JSON line to every channel in the synthetic group (killpg analog).
     * @param {number} pgid
     * @param {string} [signal]
     * @returns {{ delivered: number, channels: string[] }}
     */
    signalProcessGroup(pgid, signal) {
      const key = 'pgid:' + String(Math.floor(Number(pgid)))
      /** @type {string[]} */
      const names = []
      for (const [ch, sc] of channelScopes.entries()) {
        if (sc === key) names.push(ch)
      }
      const sig = String(signal || 'SIGTERM').trim().slice(0, 32) || 'SIGTERM'
      for (const ch of names) {
        try {
          this.pushJson(ch, {
            method: 'bareOsProcessGroupSignal',
            signal: sig,
            pgid: Math.floor(Number(pgid))
          })
        } catch {
          /* channel removed mid-iteration */
        }
      }
      return { delivered: names.length, channels: names }
    },

    /**
     * @param {string} name
     * @param {Uint8Array | ArrayBuffer} buf
     */
    push(name, buf) {
      const ch = channels.get(name)
      if (!ch) throw new Error('no such fifo: ' + name)
      if (pausedChannels.has(name)) {
        telemetry.fifoPushDenied++
        throw new Error('bare-os ipc: channel paused by soft backpressure')
      }
      try {
        ch.push(buf, limitFor(name))
        if (ch.queuedBytes >= backpressureHighWaterBytes) {
          if (!pausedChannels.has(name)) telemetry.backpressurePause++
          pausedChannels.add(name)
        }
      } catch (e) {
        if (
          e &&
          typeof e === 'object' &&
          String(/** @type {Error} */ (e).message || '').includes('backlog')
        ) {
          telemetry.fifoPushDenied++
        }
        throw e
      }
    },

    /**
     * @param {string} name
     * @param {{ signal?: AbortSignal }} [takeOpts]
     */
    take(name, takeOpts) {
      const ch = channels.get(name)
      if (!ch) return Promise.reject(new Error('no such fifo: ' + name))
      if (ch.waiters.length >= maxWaitersPerChannel) {
        telemetry.fifoTakeDeniedWaiters++
        return Promise.reject(
          new Error(
            'bare-os ipc: waiter backlog exceeds maxWaitersPerChannel'
          )
        )
      }
      return ch.take(takeOpts?.signal).catch((e) => {
        if (e && typeof e === 'object' && /** @type {Error} */ (e).name === 'AbortError') {
          telemetry.fifoTakeAborted++
        }
        throw e
      }).finally(() => {
        if (
          pausedChannels.has(name) &&
          ch.queuedBytes <= backpressureLowWaterBytes
        ) {
          pausedChannels.delete(name)
          telemetry.backpressureResume++
        }
      })
    },

    /**
     * Push a JSON-RPC-style envelope (UTF-8). Same size limits as raw `push`.
     * When `BARE_OS_IPC_RPC_TOKEN` is set on the booter, `obj.bareOsIpcToken` must match.
     * @param {string} name
     * @param {Record<string, unknown>} obj
     * @param {{ token?: string }} [pushOpts]
     */
    pushJson(name, obj, pushOpts = {}) {
      if (!obj || typeof obj !== 'object')
        throw new Error('invalid ipc json object')
      const tok =
        pushOpts.token != null && String(pushOpts.token)
          ? String(pushOpts.token)
          : rpcToken
      if (tok) {
        const got = obj.bareOsIpcToken
        if (got !== tok) throw new Error('bare-os ipc: RPC token mismatch')
      }
      const rest = { ...obj }
      delete rest.bareOsIpcToken
      const line = JSON.stringify({ bareOsRpc: '1', ...rest }) + '\n'
      if (line.length > maxJsonLine) {
        throw new Error('bare-os ipc: JSON-RPC line exceeds max length')
      }
      this.push(name, b4a.from(line, 'utf8'))
    },

    /**
     * @param {string} name
     * @returns {Promise<Record<string, unknown>>}
     */
    async takeJson(name) {
      const u8 = await this.take(name, undefined)
      const line = b4a.toString(u8, 'utf8').trim()
      if (line.length > maxJsonLine)
        throw new Error('bare-os ipc: JSON line too large')
      const j = JSON.parse(line)
      if (!j || typeof j !== 'object') throw new Error('invalid ipc json')
      if (
        typeof j.id === 'string' &&
        (Object.prototype.hasOwnProperty.call(j, 'result') ||
          Object.prototype.hasOwnProperty.call(j, 'error'))
      ) {
        const pending = pendingRequests.get(j.id)
        if (pending) {
          pendingRequests.delete(j.id)
          clearTimeout(pending.to)
          telemetry.rpcResponses++
          pending.resolve(/** @type {Record<string, unknown>} */ (j))
        } else {
          telemetry.rpcLateResponses++
        }
      }
      return /** @type {Record<string, unknown>} */ (j)
    },

    /**
     * JSON-RPC request helper with deadline and correlation id.
     * @param {string} name
     * @param {Record<string, unknown>} req
     * @param {{ timeoutMs?: number, token?: string }} [opts]
     */
    request(name, req, opts = {}) {
      const method = String(req && req.method ? req.method : 'unknown')
        .trim()
        .slice(0, 64) || 'unknown'
      const br = rpcBreakerByMethod.get(method)
      if (br && br.openUntilMs > Date.now()) {
        return Promise.reject(
          new Error('bare-os ipc: circuit open for method ' + method)
        )
      }
      const id = `rpc-${Date.now()}-${nextRequestId++}`
      const timeoutMs = Math.max(
        10,
        Math.min(120000, Math.floor(Number(opts.timeoutMs) || 8000))
      )
      telemetry.rpcRequests++
      return new Promise((resolve, reject) => {
        const to = setTimeout(() => {
          pendingRequests.delete(id)
          telemetry.rpcTimeouts++
          const row =
            rpcBreakerByMethod.get(method) ||
            { failures: 0, total: 0, openUntilMs: 0 }
          row.failures++
          row.total++
          if (row.total >= 10) {
            const ratio = row.failures / Math.max(1, row.total)
            if (ratio >= 0.5) row.openUntilMs = Date.now() + 5000
          }
          rpcBreakerByMethod.set(method, row)
          reject(new Error('bare-os ipc: request timeout'))
        }, timeoutMs)
        pendingRequests.set(id, {
          resolve: /** @type {(v: Record<string, unknown>) => void} */ (resolve),
          reject: /** @type {(e: Error) => void} */ (reject),
          to
        })
        this.pushJson(name, { ...req, id }, { token: opts.token })
      })
    },

    /**
     * @param {string} name
     * @param {string} id
     * @param {{ result?: unknown, error?: unknown, token?: string }} [payload]
     */
    respond(name, id, payload = {}) {
      const idStr = String(id || '')
      const pending = pendingRequests.get(idStr)
      if (pending) {
        pendingRequests.delete(idStr)
        clearTimeout(pending.to)
        telemetry.rpcResponses++
        pending.resolve(
          /** @type {Record<string, unknown>} */ ({
            id: idStr,
            ...(Object.prototype.hasOwnProperty.call(payload, 'error')
              ? { error: payload.error }
              : { result: payload.result })
          })
        )
        return
      }
      this.pushJson(
        name,
        {
          id: idStr,
          ...(Object.prototype.hasOwnProperty.call(payload, 'error')
            ? { error: payload.error }
            : { result: payload.result })
        },
        { token: payload.token }
      )
    },

    /**
     * Batch JSON sends in-callsite order.
     * @param {string} name
     * @param {Record<string, unknown>[]} rows
     * @param {{ token?: string }} [opts]
     */
    pushJsonBatch(name, rows, opts = {}) {
      if (!Array.isArray(rows) || rows.length === 0) return
      telemetry.jsonBatchSends++
      for (const row of rows) this.pushJson(name, row, opts)
    },

    /**
     * @param {string} name
     */
    waitUntilReady(name) {
      const row = channelReady.get(name)
      if (!row || row.ready) return Promise.resolve(true)
      return new Promise((resolve) =>
        row.waiters.push(/** @type {(v: true) => void} */ (resolve))
      )
    },

    /**
     * Publish one payload to all fan-out subscribers for `name` (copy per subscriber).
     * No-op when there are no subscribers.
     * @param {string} name
     * @param {Uint8Array | ArrayBuffer} buf
     */
    fanoutPublish(name, buf) {
      if (!fanoutOn) throw new Error('bare-os ipc: fan-out disabled')
      assertSafeIpcName(name)
      const u8 = buf instanceof Uint8Array ? buf : new Uint8Array(buf)
      if (u8.byteLength > maxFifoBytes) {
        throw new Error('fan-out payload exceeds BARE_OS_IPC_MAX_BYTES')
      }
      let hub = fanouts.get(name)
      if (!hub || hub.subscribers.length === 0) return
      const copy = new Uint8Array(u8)
      telemetry.fanoutDrop += Number(hub.publish(copy) || 0)
    },

    /**
     * Subscribe to fan-out channel `name`. Call `dispose()` when done.
     * @param {string} name
     * @returns {{ take: () => Promise<Uint8Array>, dispose: () => void }}
     */
    fanoutSubscribe(name) {
      if (!fanoutOn) throw new Error('bare-os ipc: fan-out disabled')
      assertSafeIpcName(name)
      let hub = fanouts.get(name)
      if (!hub) {
        hub = new FanoutHub(limitFor(name))
        fanouts.set(name, hub)
      }
      return hub.subscribe()
    },

    /**
     * @param {string} name
     */
    fanoutSubscriberCount(name) {
      const hub = fanouts.get(name)
      return hub ? hub.subscribers.length : 0
    },

    /**
     * POSIX-like named message queue (separate from FIFO byte streams).
     * @param {string} name
     * @param {{ maxmsg?: number, maxBytes?: number }} [mqOpts]
     */
    mqOpen(name, mqOpts = {}) {
      assertSafeIpcName(name)
      if (mqQueues.has(name)) {
        const q = mqQueues.get(name)
        return {
          name,
          maxmsg: q.maxmsg,
          maxBytes: q.maxBytes,
          curmsgs: q.msgs.length
        }
      }
      const optMax = mqOpts.maxmsg
      const maxmsg = Math.min(
        mqMsgCeiling,
        Math.max(
          1,
          Math.floor(
            Number.isFinite(Number(optMax)) && Number(optMax) >= 1
              ? Number(optMax)
              : mqDefaultMaxmsg
          )
        )
      )
      const optBytes = mqOpts.maxBytes
      const maxBytes = Math.min(
        1024 * 1024,
        Math.max(
          256,
          Math.floor(
            Number.isFinite(Number(optBytes)) && Number(optBytes) >= 256
              ? Number(optBytes)
              : mqDefaultMaxBytes
          )
        )
      )
      mqQueues.set(name, {
        maxmsg,
        maxBytes,
        nextSeq: 0,
        msgs: [],
        curBytes: 0
      })
      return { name, maxmsg, maxBytes, curmsgs: 0 }
    },

    /** @param {string} name */
    mqAttrs(name) {
      const q = mqQueues.get(name)
      if (!q) return null
      return {
        maxmsg: q.maxmsg,
        maxBytes: q.maxBytes,
        curmsgs: q.msgs.length,
        curBytes: q.curBytes
      }
    },

    /**
     * @param {string} name
     * @param {number} prio
     * @param {Uint8Array | string} data
     */
    mqSend(name, prio, data) {
      const q = mqQueues.get(name)
      if (!q) throw new Error('bare-os ipc mq: unknown queue (mqOpen first)')
      const u8 =
        typeof data === 'string' ? b4a.from(data, 'utf8') : new Uint8Array(data)
      const p = Math.max(0, Math.min(32767, Math.floor(Number(prio) || 0)))
      if (u8.byteLength > q.maxBytes) {
        throw new Error('bare-os ipc mq: message exceeds maxBytes')
      }
      if (q.msgs.length >= q.maxmsg) {
        telemetry.mqFull++
        throw new Error('bare-os ipc mq: queue full (maxmsg)')
      }
      if (q.curBytes + u8.byteLength > q.maxmsg * q.maxBytes) {
        throw new Error('bare-os ipc mq: queue byte budget exceeded')
      }
      const seq = q.nextSeq++
      const row = { prio: p, seq, data: u8 }
      let idx = q.msgs.length
      while (idx > 0) {
        const prev = q.msgs[idx - 1]
        if (prev.prio > row.prio) break
        if (prev.prio === row.prio && prev.seq < row.seq) break
        idx--
      }
      q.msgs.splice(idx, 0, row)
      q.curBytes += u8.byteLength
      return { ok: true }
    },

    /** @param {string} name */
    mqReceive(name) {
      const q = mqQueues.get(name)
      if (!q || q.msgs.length === 0) return null
      const m = q.msgs.shift()
      if (!m) return null
      q.curBytes -= m.data.byteLength
      return { prio: m.prio, data: m.data }
    },

    stats() {
      let queuedBytesTotal = 0
      /** @type {{ name: string, queuedBytes: number, waiters: number }[]} */
      const fifoDepth = []
      for (const [name, ch] of channels.entries()) {
        queuedBytesTotal += ch.queuedBytes
        fifoDepth.push({
          name,
          queuedBytes: ch.queuedBytes,
          waiters: ch.waiters.length
        })
      }
      fifoDepth.sort((a, b) => b.queuedBytes - a.queuedBytes)
      let fanoutSubscribersTotal = 0
      for (const hub of fanouts.values()) {
        fanoutSubscribersTotal += hub.subscribers.length
      }
      /** @type {Record<string, number>} */
      const scopeHistogram = {}
      for (const sc of channelScopes.values()) {
        scopeHistogram[sc] = (scopeHistogram[sc] || 0) + 1
      }
      /** @type {Record<string, string[]>} */
      const byPgid = {}
      for (const [ch, sc] of channelScopes.entries()) {
        if (!sc.startsWith('pgid:')) continue
        if (!byPgid[sc]) byPgid[sc] = []
        byPgid[sc].push(ch)
      }
      for (const k of Object.keys(byPgid)) {
        byPgid[k].sort()
      }
      return {
        channelCount: channels.size,
        maxChannels,
        telemetry: { ...telemetry },
        waiterLimits: {
          schema: 1,
          maxWaitersPerChannel,
          softPauseChannels: pausedChannels.size,
          highWaterBytes: backpressureHighWaterBytes,
          lowWaterBytes: backpressureLowWaterBytes
        },
        rpcCircuitBreaker: {
          schema: 1,
          methods: Object.fromEntries(
            [...rpcBreakerByMethod.entries()].map(([k, v]) => [
              k,
              {
                failures: v.failures,
                total: v.total,
                open: v.openUntilMs > Date.now(),
                openForMs:
                  v.openUntilMs > Date.now() ? v.openUntilMs - Date.now() : 0
              }
            ])
          )
        },
        queuedBytesTotal,
        fanoutTopicCount: fanouts.size,
        fanoutSubscribersTotal,
        channelScopes: Object.fromEntries([...channelScopes.entries()].sort()),
        scopeHistogram,
        ipcBackpressure: {
          schema: 1,
          topChannels: fifoDepth.slice(0, 32),
          note: 'Per-FIFO depth + waiter count for operator /proc mirrors.'
        },
        processGroups: {
          schema: 1,
          byPgid,
          note: 'Channels tagged via assignProcessGroup; consumers handle bareOsProcessGroupSignal RPC lines.'
        },
        posixMessageQueues: {
          schema: 2,
          count: mqQueues.size,
          names: [...mqQueues.keys()].sort().slice(0, 128),
          note: 'Separate from FIFOs; mq_receive returns highest priority first; equal priority FIFO by enqueue order (seq).'
        }
      }
    },

    /**
     * Bounded duplex: side A's `push` delivers to side B's `take`, and vice versa.
     * @param {string} baseName
     * @returns {{ left: { push: (buf: Uint8Array | ArrayBuffer) => void, take: () => Promise<Uint8Array> }, right: { push: (buf: Uint8Array | ArrayBuffer) => void, take: () => Promise<Uint8Array> } }}
     */
    createDuplexBridge(baseName) {
      assertSafeIpcName(baseName)
      const lim = limitFor(baseName)
      const ab = new FifoChannel()
      const ba = new FifoChannel()
      const left = {
        push: (buf) => ba.push(buf, lim),
        take: (sig) => ab.take(sig)
      }
      const right = {
        push: (buf) => ab.push(buf, lim),
        take: (sig) => ba.take(sig)
      }
      return { left, right }
    },

    /**
     * One JSON line request/response over a {@link createDuplexBridge} side (`bareOsRpc: "2"`).
     * @param {{ push: (buf: Uint8Array | ArrayBuffer) => void, take: () => Promise<Uint8Array> }} side
     * @param {Record<string, unknown>} request
     */
    async duplexJsonRoundTrip(side, request) {
      if (!request || typeof request !== 'object')
        throw new Error('bare-os ipc: invalid duplex request')
      const line =
        JSON.stringify({
          bareOsRpc: '2',
          ...request
        }) + '\n'
      if (line.length > maxJsonLine) {
        throw new Error('bare-os ipc: duplex JSON line exceeds max length')
      }
      side.push(b4a.from(line, 'utf8'))
      const u8 = await side.take()
      const res = b4a.toString(u8, 'utf8').trim()
      if (res.length > maxJsonLine) {
        throw new Error('bare-os ipc: duplex response too large')
      }
      return /** @type {Record<string, unknown>} */ (JSON.parse(res))
    }
  }
}