Class SocketTransport

src/mcp/transport.ts:344–436

 * ending must not bring down the whole daemon.
 */
export class SocketTransport extends LineBasedJsonRpcTransport {
  private buffer = '';
  private closeHandlers: Array<() => void> = [];

  constructor(private socket: Socket, private prefix: string = 'cg-sock') {
    super();
  }

  /**
   * Register a callback fired exactly once when the socket closes (from either
   * side). Used by the daemon to decrement its connected-clients refcount.
   */
  onClose(handler: () => void): void {
    this.closeHandlers.push(handler);
  }

  start(handler: MessageHandler): void {
    this.messageHandler = handler;

    this.socket.setEncoding('utf8');
    if (process.env.CODEGRAPH_MCP_DEBUG) {
      process.stderr.write(`[mcp-debug] transport attached flowing=${String(this.socket.readableFlowing)} buffered=${this.socket.readableLength}\n`);
    }
    this.socket.on('data', (chunk: string) => {
      if (process.env.CODEGRAPH_MCP_DEBUG) process.stderr.write(`[mcp-debug] transport data ${chunk.length}b\n`);
      this.buffer += chunk;
      let idx;
      // Drain every complete line; tail-fragment stays in the buffer for the
      // next chunk. The handler is async but we don't await it here — JSON-RPC
      // permits out-of-order responses, and serializing here would deadlock if
      // a handler issued a server-initiated request that needed a *later* line
      // to arrive (e.g. roots/list mid-tools-call).
      while ((idx = this.buffer.indexOf('\n')) !== -1) {
        const line = this.buffer.slice(0, idx);
        this.buffer = this.buffer.slice(idx + 1);
        void this.handleLine(line);
      }
    });

    this.socket.on('close', () => this.handleSocketClose());
    this.socket.on('error', (err) => {
      // Don't crash the daemon over a broken pipe; just shut this connection.
      process.stderr.write(`[CodeGraph daemon] socket error: ${err.message}\n`);
      this.handleSocketClose();
    });
    // The daemon's hello reader hands the socket over PAUSED (so the unshifted
    // tail can't be emitted to zero listeners and lost — the #662 wedge).
    // Attaching 'data' does not resume an explicitly-paused stream; do it here.
    // Harmless when the socket was never paused.
    this.socket.resume();
  }

  stop(): void {
    if (this.stopped) return;
    this.stopped = true;
    this.rejectPending('Transport stopped');
    if (!this.socket.destroyed) {
      this.socket.end();