spb/claude-remote

1/**2 * Client for apps/runtime over its 0600 unix socket (HTTP + WebSocket, bearer token).3 * The server never spawns processes or touches project files itself: everything goes through here.4 */5import type { Readable } from 'node:stream';6import { Agent, request, type Dispatcher } from 'undici';7import WebSocket from 'ws';8import { RtEvent } from '@cr/shared';910export class RuntimeError extends Error {11  constructor(12    /** HTTP status from the runtime; 0 = unreachable (socket missing, refused, timeout). */13    public readonly status: number,14    public readonly code: string,15    message: string,16    /** `error.details` of the runtime body (e.g. `{sha256, mtimeMs}` of a write conflict). */17    public readonly details?: unknown,18  ) {19    super(message);20  }21  get unreachable(): boolean {22    return this.status === 0;23  }24}2526export interface RawResponse {27  statusCode: number;28  headers: Record<string, string | string[] | undefined>;29  body: Dispatcher.ResponseData['body'];30}31type Query = Record<string, string | number | undefined>;3233export class RuntimeClient {34  private readonly agent: Agent;35  private closed = false;3637  constructor(38    public readonly socketPath: string,39    private readonly token: string,40  ) {41    this.agent = new Agent({ connect: { socketPath }, connections: 32, keepAliveTimeout: 10_000 });42  }4344  private url(path: string, query?: Query): string {45    const q = new URLSearchParams();46    for (const [k, v] of Object.entries(query ?? {})) if (v !== undefined) q.set(k, String(v));47    const qs = q.toString();48    return `http://runtime${path}${qs ? `?${qs}` : ''}`;49  }5051  /** Streaming request. Non-2xx responses are converted to RuntimeError (body drained). */52  async raw(53    method: string,54    path: string,55    opts: { query?: Query; body?: Readable | string | Buffer; headers?: Record<string, string>; timeoutMs?: number } = {},56  ): Promise<RawResponse> {57    let res;58    try {59      res = await request(this.url(path, opts.query), {60        dispatcher: this.agent,61        method: method as 'GET',62        body: opts.body,63        headers: { authorization: `Bearer ${this.token}`, ...opts.headers },64        headersTimeout: opts.timeoutMs ?? 30_000,65        bodyTimeout: 0,66      });67    } catch (err) {68      throw new RuntimeError(0, 'runtime_unavailable', `runtime unreachable: ${(err as NodeJS.ErrnoException).code ?? 'error'}`);69    }70    if (res.statusCode >= 400) {71      let code = 'runtime_error';72      let message = `runtime responded ${res.statusCode}`;73      let details: unknown;74      try {75        const parsed = (await res.body.json()) as { error?: { code?: unknown; message?: unknown; details?: unknown } };76        if (typeof parsed.error?.code === 'string') code = parsed.error.code;77        if (typeof parsed.error?.message === 'string') message = parsed.error.message;78        details = parsed.error?.details;79      } catch {80        /* non-JSON error body */81      }82      throw new RuntimeError(res.statusCode, code, message, details);83    }84    return { statusCode: res.statusCode, headers: res.headers, body: res.body };85  }8687  async json<T>(method: string, path: string, body?: unknown, opts: { query?: Query; timeoutMs?: number } = {}): Promise<T> {88    const res = await this.raw(method, path, {89      query: opts.query,90      timeoutMs: opts.timeoutMs,91      body: body === undefined ? undefined : JSON.stringify(body),92      headers: body === undefined ? {} : { 'content-type': 'application/json' },93    });94    try {95      const text = await res.body.text();96      return (text ? JSON.parse(text) : null) as T;97    } catch {98      throw new RuntimeError(502, 'runtime_bad_response', 'runtime returned invalid JSON');99    }100  }101102  /** WebSocket to the runtime on the same unix socket. */103  ws(path: string, query?: Query, maxPayload = 16 * 1024 * 1024): WebSocket {104    const u = this.url(path, query).replace('http://runtime', '');105    return new WebSocket(`ws+unix://${this.socketPath}:${u}`, {106      headers: { authorization: `Bearer ${this.token}` },107      maxPayload,108      perMessageDeflate: false,109      handshakeTimeout: 10_000,110    });111  }112113  /** Reconnecting `/events` subscriber (1 s → 10 s backoff). `onState` reports link up/down. */114  subscribeEvents(onEvent: (ev: RtEvent) => void, onState?: (connected: boolean) => void): { close(): void } {115    let stopped = false;116    let socket: WebSocket | null = null;117    let timer: NodeJS.Timeout | null = null;118    let delay = 1000;119    const connect = (): void => {120      if (stopped || this.closed) return;121      const ws = this.ws('/events');122      socket = ws;123      ws.on('open', () => {124        delay = 1000;125        onState?.(true);126      });127      ws.on('message', (data, isBinary) => {128        if (isBinary) return;129        try {130          const ev = RtEvent.safeParse(JSON.parse(data.toString()));131          if (ev.success) onEvent(ev.data);132        } catch {133          /* ignore malformed frames */134        }135      });136      ws.on('error', () => undefined);137      ws.on('close', () => {138        onState?.(false);139        if (stopped || this.closed) return;140        timer = setTimeout(connect, delay);141        timer.unref();142        delay = Math.min(delay * 2, 10_000);143      });144    };145    connect();146    return {147      close: () => {148        stopped = true;149        if (timer) clearTimeout(timer);150        socket?.terminate();151      },152    };153  }154155  async close(): Promise<void> {156    this.closed = true;157    await this.agent.close().catch(() => undefined);158  }159}160