Repository Public
1/**2 * Test-only helpers for the runtime-core suites (real tmux, real unix socket, temp dirs).3 * Imported exclusively by `*.test.ts`; nothing in the runtime depends on it.4 */5import crypto from 'node:crypto';6import fs from 'node:fs';7import http from 'node:http';8import path from 'node:path';9import { execFileSync } from 'node:child_process';10import WebSocket from 'ws';11import type { RtEvent, TermServerMsg } from '@cr/shared';12import type { RuntimeConfig } from './config.js';13import { buildRuntime, type Runtime } from './main.js';1415export interface TestEnv {16 dir: string;17 workspace: string;18 config: RuntimeConfig;19 /** Kills the dedicated tmux server and removes the temp dir. */20 cleanup(): void;21}2223export function makeTestEnv(overrides: Partial<RuntimeConfig> = {}): TestEnv {24 // /tmp keeps the unix socket path well under the 104-byte limit of macOS.25 const dir = fs.realpathSync(fs.mkdtempSync('/tmp/crt-'));26 fs.chmodSync(dir, 0o700);27 const workspace = path.join(dir, 'ws');28 fs.mkdirSync(workspace);29 const dataDir = path.join(dir, 'data');30 const config: RuntimeConfig = {31 dataDir,32 socketPath: path.join(dataDir, 'run', 'runtime.sock'),33 token: crypto.randomBytes(32).toString('hex'),34 workspaceRoots: [{ label: 'ws', path: workspace }],35 agent: 'claude',36 claudeBin: 'claude',37 maxSessions: 12,38 execHosts: [],39 termBufferBytes: 4 * 1024 * 1024,40 tmuxSocketName: `crt-${crypto.randomBytes(6).toString('hex')}`,41 shell: '/bin/sh',42 logLevel: 'silent',43 ...overrides,44 };45 return {46 dir,47 workspace,48 config,49 cleanup() {50 try {51 execFileSync('tmux', ['-L', config.tmuxSocketName, 'kill-server'], { stdio: 'ignore' });52 } catch {53 /* no server left */54 }55 // tmux leaves the socket file of a killed server behind.56 const uid = typeof process.getuid === 'function' ? process.getuid() : 0;57 fs.rmSync(path.join(process.env.TMUX_TMPDIR ?? '/tmp', `tmux-${uid}`, config.tmuxSocketName), { force: true });58 fs.rmSync(dir, { recursive: true, force: true });59 },60 };61}6263export async function startRuntime(config: RuntimeConfig): Promise<Runtime> {64 const rt = await buildRuntime(config, { plugins: 'core' });65 await rt.start();66 return rt;67}6869export interface HttpResult {70 status: number;71 headers: http.IncomingHttpHeaders;72 body: Buffer;73 // Test convenience: callers narrow the parsed JSON themselves.74 json<T = unknown>(): T;75}7677export function request(78 socketPath: string,79 method: string,80 urlPath: string,81 opts: { token?: string | null; body?: unknown; rawBody?: string; headers?: Record<string, string> } = {},82): Promise<HttpResult> {83 return new Promise((resolve, reject) => {84 const payload = opts.rawBody !== undefined ? Buffer.from(opts.rawBody) : opts.body === undefined ? null : Buffer.from(JSON.stringify(opts.body));85 const headers: Record<string, string> = { ...opts.headers };86 if (opts.token) headers.authorization = `Bearer ${opts.token}`;87 if (payload) {88 headers['content-type'] ??= 'application/json';89 headers['content-length'] = String(payload.length);90 }91 const req = http.request({ socketPath, method, path: urlPath, headers }, (res) => {92 const chunks: Buffer[] = [];93 res.on('data', (c: Buffer) => chunks.push(c));94 res.on('end', () => {95 const body = Buffer.concat(chunks);96 resolve({ status: res.statusCode ?? 0, headers: res.headers, body, json: <T>() => JSON.parse(body.toString('utf8')) as T });97 });98 });99 req.on('error', reject);100 if (payload) req.write(payload);101 req.end();102 });103}104105export async function waitUntil<T>(fn: () => T | Promise<T>, timeoutMs = 10_000, what = 'condition'): Promise<NonNullable<T>> {106 const deadline = Date.now() + timeoutMs;107 for (;;) {108 const v = await fn();109 if (v) return v as NonNullable<T>;110 if (Date.now() > deadline) throw new Error(`timed out waiting for ${what}`);111 await new Promise((r) => setTimeout(r, 50));112 }113}114115/** Opens a WebSocket over the unix socket. Rejects with `HTTP <status>` when the upgrade is refused. */116export function openWs(socketPath: string, urlPath: string, token: string | null): Promise<WebSocket> {117 return new Promise((resolve, reject) => {118 const ws = new WebSocket(`ws+unix:${socketPath}:${urlPath}`, { headers: token ? { authorization: `Bearer ${token}` } : {} });119 ws.once('open', () => resolve(ws));120 ws.once('unexpected-response', (_req, res) => {121 res.resume();122 reject(new Error(`HTTP ${res.statusCode}`));123 });124 ws.once('error', reject);125 });126}127128/** Terminal WS client: accumulates binary output and control messages. */129export class TermClient {130 output = '';131 bytes = 0;132 readonly msgs: TermServerMsg[] = [];133 closeCode: number | null = null;134135 private constructor(readonly ws: WebSocket) {136 ws.on('message', (data: WebSocket.RawData, isBinary: boolean) => {137 const buf = Buffer.isBuffer(data) ? data : Array.isArray(data) ? Buffer.concat(data) : Buffer.from(data);138 if (isBinary) {139 this.bytes += buf.length;140 this.output += buf.toString('utf8');141 } else this.msgs.push(JSON.parse(buf.toString('utf8')) as TermServerMsg);142 });143 ws.on('close', (code) => {144 this.closeCode = code;145 });146 }147148 static async open(socketPath: string, sessionId: string, token: string, cols = 100, rows = 30): Promise<TermClient> {149 const ws = new WebSocket(`ws+unix:${socketPath}:/sessions/${sessionId}/attach?cols=${cols}&rows=${rows}`, {150 headers: { authorization: `Bearer ${token}` },151 });152 const client = new TermClient(ws); // listeners attached before 'open' so no frame is missed153 await new Promise<void>((resolve, reject) => {154 ws.once('open', () => resolve());155 ws.once('error', reject);156 });157 return client;158 }159160 type(text: string): void {161 this.ws.send(Buffer.from(text, 'utf8'), { binary: true });162 }163 control(msg: unknown): void {164 this.ws.send(JSON.stringify(msg));165 }166 waitOutput(needle: string | RegExp, timeoutMs = 10_000): Promise<boolean> {167 return waitUntil(() => (typeof needle === 'string' ? this.output.includes(needle) : needle.test(this.output)), timeoutMs, `output ${String(needle)}`);168 }169 waitMsg<K extends TermServerMsg['t']>(t: K, timeoutMs = 10_000): Promise<Extract<TermServerMsg, { t: K }>> {170 return waitUntil(() => this.msgs.find((m): m is Extract<TermServerMsg, { t: K }> => m.t === t), timeoutMs, `message ${t}`);171 }172 waitClose(timeoutMs = 10_000): Promise<number> {173 return waitUntil(() => this.closeCode, timeoutMs, 'ws close');174 }175 async close(): Promise<void> {176 if (this.closeCode === null) {177 this.ws.close();178 await this.waitClose();179 }180 }181}182183/** Collects `RtEvent`s from WS /events. */184export async function collectEvents(socketPath: string, token: string): Promise<{ events: RtEvent[]; close(): void }> {185 const ws = await openWs(socketPath, '/events', token);186 const events: RtEvent[] = [];187 ws.on('message', (data: WebSocket.RawData) => events.push(JSON.parse(String(data)) as RtEvent));188 return { events, close: () => ws.close() };189}190191export function sessionSpec(env: TestEnv, extra: Record<string, unknown> = {}): Record<string, unknown> {192 return { id: crypto.randomUUID(), kind: 'shell', cwd: env.workspace, args: [], env: {}, cols: 100, rows: 30, ...extra };193}194195export function taskSpec(env: TestEnv, command: string, extra: Record<string, unknown> = {}): Record<string, unknown> {196 return { id: crypto.randomUUID(), command, cwd: env.workspace, host: 'local', timeoutSec: null, env: {}, ...extra };197}198