Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288/** * Unix-socket server test — saga tb4593.3 (PROTOCOL §7.1, §9.1 hello-on-connect). * * Connects a REAL `net.createConnection` client to a `SocketServer` over a tmp socket * path. Hermetic: tmp dir, no network, no daemon, no agent. Verifies the framing reuse * (LF-only JSONL, U+2028/U+2029-safe) and the hello-on-connect handshake against the * `HelloSchema` shape. * * Coverage: * - start() binds; a client connects; `hello` arrives first (§9.1) + validates schema. * - onConnect fires with a Connection; `send()` writes JSONL to the client. * - `messages()` yields inbound JSONL records; a multi-line chunk splits correctly. * - a partial record split across chunks reassembles (streaming JsonlDecoder). * - a U+2028 inside a JSON string survives intact (the readline bug, §2.1). * - a malformed JSON line yields a record with `parseError` (no stream abort, §8). * - stale-socket removal at boot (a leftover file at the path doesn't block bind). * - stop() unlinks the socket + destroys connections; clients observe EOF. * - connections() async iterator yields accepted connections. */import { describe, expect, it, beforeEach, afterEach } from 'bun:test';import { mkdtempSync, rmSync, writeFileSync, existsSync } from 'node:fs';import { join } from 'node:path';import { tmpdir } from 'node:os';import { createConnection } from 'node:net';import { Value } from '@sinclair/typebox/value';import { HelloSchema } from '@/protocol/lifecycle';import { encodeJsonl, splitJsonl } from '@/protocol/framing/jsonl';import { SocketServer, type Connection } from '@/daemon/socket-server';
let dir: string;let sockPath: string;
beforeEach(() => { dir = mkdtempSync(join(tmpdir(), 'shard-sock-')); sockPath = join(dir, 'daemon.sock');});
afterEach(() => { rmSync(dir, { recursive: true, force: true });});
/** Promise-resolved on the next `length` LF-delimited records from a socket. */function readN(socket: NodeJS.ReadableStream, n: number): Promise<string[]> { return new Promise((resolve) => { const lines: string[] = []; let buf = ''; const onData = (chunk: Buffer) => { buf += chunk.toString('utf8'); while (buf.includes('\n')) { const nl = buf.indexOf('\n'); const line = buf.slice(0, nl); if (line.length > 0) lines.push(line); buf = buf.slice(nl + 1); } if (lines.length >= n) { socket.off('data', onData); resolve(lines); } }; socket.on('data', onData); });}
/** Connect a raw client + collect the first record (the hello). */async function connectAndHello(server: SocketServer): Promise<{ sock: ReturnType<typeof createConnection>; hello: unknown;}> { const sock = createConnection({ path: server.path }); // Wait for the socket to be writable. await new Promise<void>((resolve) => sock.once('connect', () => resolve())); const [helloLine] = await readN(sock, 1).then((lines) => [JSON.parse(lines[0]!)]); return { sock, hello: helloLine };}describe('SocketServer — hello + framing (tb4593.3)', () => { it('binds and emits hello on connect matching HelloSchema (§9.1)', async () => { const server = new SocketServer({ socketPath: sockPath }); await server.start();
const { hello } = await connectAndHello(server); expect(Value.Check(HelloSchema, hello)).toBe(true); expect((hello as { type: string }).type).toBe('hello'); expect((hello as { server_version: string }).server_version).toContain('shard');
await server.stop(); });
it('onConnect fires with a Connection after hello is sent', async () => { const server = new SocketServer({ socketPath: sockPath }); await server.start(); const seen: Connection[] = []; server.onConnect((conn) => seen.push(conn));
await connectAndHello(server); await new Promise((r) => setTimeout(r, 10)); // microtask flush
expect(seen).toHaveLength(1); expect(seen[0]?.closed).toBe(false);
await server.stop(); });
it('drops closed clients from live connection tracking', async () => { const server = new SocketServer({ socketPath: sockPath }); await server.start(); const { sock } = await connectAndHello(server); expect(server.liveConnections()).toHaveLength(1);
const closed = new Promise<void>((resolve) => sock.once('close', () => resolve())); sock.destroy(); await closed; await new Promise((resolve) => setTimeout(resolve, 10)); expect(server.liveConnections()).toHaveLength(0);
await server.stop(); });
it('send() writes a JSONL message to the client', async () => { const server = new SocketServer({ socketPath: sockPath }); await server.start(); let conn: Connection | undefined; server.onConnect((c) => { conn = c; });
const { sock } = await connectAndHello(server); await new Promise((r) => setTimeout(r, 10));
await conn!.send({ type: 'status', status: 'idle' }); const [line] = await readN(sock, 1).then((ls) => [JSON.parse(ls[0]!)]); expect(line).toEqual({ type: 'status', status: 'idle' });
await server.stop(); });
it('messages() yields inbound JSONL records (multi-line chunk splits)', async () => { const server = new SocketServer({ socketPath: sockPath }); await server.start(); let conn: Connection | undefined; server.onConnect((c) => { conn = c; }); const { sock } = await connectAndHello(server); await new Promise((r) => setTimeout(r, 10));
const iter = conn!.messages(); // Send two records in one write — JsonlDecoder splits on \n. sock.write(encodeJsonl({ type: 'agent.list' }) + encodeJsonl({ type: 'agent.status' }));
const r1 = (await iter.next()).value; const r2 = (await iter.next()).value; expect(r1?.value).toEqual({ type: 'agent.list' }); expect(r2?.value).toEqual({ type: 'agent.status' });
await server.stop(); });
it('reassembles a record split across chunks (streaming decoder)', async () => { const server = new SocketServer({ socketPath: sockPath }); await server.start(); let conn: Connection | undefined; server.onConnect((c) => { conn = c; }); const { sock } = await connectAndHello(server); await new Promise((r) => setTimeout(r, 10));
const iter = conn!.messages(); // Write the first half, then the second half + terminator — one record total. const full = encodeJsonl({ type: 'prompt', message: 'hello world' }); const mid = Math.floor(full.length / 2); sock.write(full.slice(0, mid)); sock.write(full.slice(mid));
const r = (await iter.next()).value; expect(r?.value).toEqual({ type: 'prompt', message: 'hello world' });
await server.stop(); });
it('survives U+2028 inside a JSON string (§2.1 — the readline bug)', async () => { const server = new SocketServer({ socketPath: sockPath }); await server.start(); let conn: Connection | undefined; server.onConnect((c) => { conn = c; }); const { sock } = await connectAndHello(server); await new Promise((r) => setTimeout(r, 10));
const iter = conn!.messages(); const payload = { type: 'prompt', message: 'line\u2028break' }; // U+2028 inside the string sock.write(encodeJsonl(payload));
const r = (await iter.next()).value; expect(r?.value).toEqual(payload); // `splitJsonl` confirms a single line — readline would have split it into two. expect(splitJsonl(encodeJsonl(payload))).toHaveLength(1);
await server.stop(); });
it('yields a record with parseError on malformed JSON (no stream abort)', async () => { const server = new SocketServer({ socketPath: sockPath }); await server.start(); let conn: Connection | undefined; server.onConnect((c) => { conn = c; }); const { sock } = await connectAndHello(server); await new Promise((r) => setTimeout(r, 10));
const iter = conn!.messages(); sock.write('{ not valid json }\n');
const r = (await iter.next()).value; expect(r?.value).toBeUndefined(); expect(typeof r?.parseError).toBe('string');
await server.stop(); });
it('removes a stale socket file at boot (does not block bind)', async () => { // Leave a stale file at the socket path — a crashed-daemon leftover. writeFileSync(sockPath, 'stale'); expect(existsSync(sockPath)).toBe(true);
const server = new SocketServer({ socketPath: sockPath }); await server.start(); // unlinks stale + binds expect(existsSync(sockPath)).toBe(true); // now a real socket, not the stale file
const { hello } = await connectAndHello(server); expect((hello as { type: string }).type).toBe('hello');
await server.stop(); });
it('refuses to unlink and replace a live daemon socket', async () => { const first = new SocketServer({ socketPath: sockPath }); const second = new SocketServer({ socketPath: sockPath }); await first.start();
await expect(second.start()).rejects.toMatchObject({ code: 'EADDRINUSE' });
// The failed second startup must leave the original pathname and listener intact. const { hello } = await connectAndHello(first); expect((hello as { type: string }).type).toBe('hello'); await first.stop(); });
it('stop() unlinks the socket file and destroys live connections', async () => { const server = new SocketServer({ socketPath: sockPath }); await server.start(); const { sock } = await connectAndHello(server); await new Promise((r) => setTimeout(r, 10));
await server.stop();
// The socket file is gone. expect(existsSync(sockPath)).toBe(false); // The client observed EOF/closed. await new Promise<void>((resolve) => { if (sock.destroyed) return resolve(); sock.once('close', () => resolve()); }); expect(sock.destroyed).toBe(true); });
it('connections() async iterator yields accepted connections', async () => { const server = new SocketServer({ socketPath: sockPath }); await server.start(); const it = server.connections();
// Connect a client; the iterator should produce it. const connectP = new Promise<void>((resolve) => { const sock = createConnection({ path: server.path }); sock.once('connect', () => resolve()); void sock; }); const connP = it.next(); await connectP; const { value: conn } = await connP; expect(conn).toBeDefined(); expect(conn?.closed).toBe(false);
await server.stop(); });});