boocode/apps/coder/src/services/acp-stream.ts

import { Readable, Writable } from 'node:stream';
import type { ChildProcess } from 'node:child_process';
import { ndJsonStream } from '@agentclientprotocol/sdk';

export function nodeReadableToWeb(nodeStream: NodeJS.ReadableStream): ReadableStream<Uint8Array> {
  return new ReadableStream<Uint8Array>({
    start(controller) {
      nodeStream.on('data', (chunk: Buffer) => controller.enqueue(new Uint8Array(chunk)));
      nodeStream.on('end', () => controller.close());
      nodeStream.on('error', (err) => controller.error(err));
    },
    cancel() {
      if ('destroy' in nodeStream && typeof (nodeStream as Readable).destroy === 'function') {
        (nodeStream as Readable).destroy();
      }
    },
  });
}

export function nodeWritableToWeb(nodeStream: NodeJS.WritableStream): WritableStream<Uint8Array> {
  return new WritableStream<Uint8Array>({
    write(chunk) {
      return new Promise<void>((resolve, reject) => {
        const ok = (nodeStream as Writable).write(chunk, (err) => {
          if (err) reject(err);
        });
        if (ok) resolve();
        else (nodeStream as Writable).once('drain', resolve);
      });
    },
    close() {
      return new Promise<void>((resolve) => {
        (nodeStream as Writable).end(resolve);
      });
    },
    abort() {
      (nodeStream as Writable).destroy();
    },
  });
}

export function createAcpNdJsonStream(child: ChildProcess) {
  return ndJsonStream(nodeWritableToWeb(child.stdin!), nodeReadableToWeb(child.stdout!));
}