/** * @fileoverview Integration tests for the TUI's live-update stream. * * These run against a real loopback `text/event-stream` endpoint rather than a * mocked socket, because the behaviours that matter here are all socket-level: * a stream that ENDS, a stream that goes SILENT without erroring (the failure * mode `EventSource` cannot see, which is why the server heartbeats), and a * teardown that must leave no timer behind. */ import { describe, it, expect, beforeAll, afterAll, beforeEach, afterEach } from 'vitest'; import http from 'node:http'; import { TuiClient, type TuiApprovalEvent, type TuiSseStatusDetail } from '../../src/tui/tui-client.js'; const PORT = 3242; const BASE_URL = `http://127.0.0.1:${PORT}`; interface Connection { url: string; headers: http.IncomingHttpHeaders; res: http.ServerResponse; } const connections: Connection[] = []; /** Flipped by a test that wants every connect attempt to fail. */ let refuse = false; let server: http.Server; let client: TuiClient | null = null; function frame(event: string, data: unknown): string { return `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`; } async function until(predicate: () => boolean, timeoutMs = 3000): Promise { const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { if (predicate()) return; await new Promise((resolve) => setTimeout(resolve, 10)); } throw new Error('condition not met before the deadline'); } beforeAll(async () => { server = http.createServer((req, res) => { if (!req.url?.startsWith('/api/events')) { res.writeHead(404).end(); return; } if (refuse) { res.writeHead(503).end('busy'); return; } res.writeHead(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache' }); // Node holds headers back until the first body write; the real server sends // an `init` frame immediately, so flush to match it. Without this the // client never sees a response and every test here waits forever. res.flushHeaders(); connections.push({ url: req.url, headers: req.headers, res }); }); await new Promise((resolve) => server.listen(PORT, '127.0.0.1', resolve)); }); afterAll(async () => { await new Promise((resolve) => server.close(() => resolve())); }); beforeEach(() => { connections.length = 0; refuse = false; }); afterEach(() => { client?.close(); client = null; for (const connection of connections) connection.res.end(); }); describe('subscribeEvents', () => { it('routes each frame to the handler that owns it', async () => { const resyncs: string[] = []; const approvals: TuiApprovalEvent[] = []; let planUsage: unknown = null; let init: unknown = null; client = new TuiClient({ baseUrl: BASE_URL, password: 's3cret' }); client.subscribeEvents({ onInit: (state) => { init = state; }, onResync: (event) => resyncs.push(event), onApproval: (event) => approvals.push(event), onPlanUsage: (usage) => { planUsage = usage; }, }); await until(() => connections.length === 1); const { res } = connections[0]; res.write(frame('init', { version: '9.9.9', planUsage: { fiveHour: { usedPercentage: 5, resetAt: 1 } } })); res.write(frame('session:created', { id: 'a' })); // Split across writes on purpose: the parser must not need frame-aligned reads. res.write('event: approval:pending\ndata: {"id":"a:1","sessionId":"a",'); res.write('"kind":"permission","createdAt":7}\n\n'); res.write(frame('session:terminal', { id: 'a', data: 'noise' })); res.write(frame('sse:heartbeat', { t: 1 })); res.write(frame('session:statusTelemetry', { sessionId: 'a', fiveHour: { usedPercentage: 41, resetAt: 2 } })); await until(() => planUsage !== null); expect(init).toEqual({ version: '9.9.9', planUsage: { fiveHour: { usedPercentage: 5, resetAt: 1 } } }); expect(approvals).toEqual([ { kind: 'pending', item: { id: 'a:1', sessionId: 'a', kind: 'permission', createdAt: 7 } }, ]); expect(planUsage).toEqual({ sessionId: 'a', fiveHour: { usedPercentage: 41, resetAt: 2 } }); // The approval also regrouped a row, so it resyncs too. Terminal and // heartbeat frames never do. expect(resyncs).toEqual(['session:created', 'approval:pending']); }); it('suppresses the terminal firehose by default and carries the auth header', async () => { client = new TuiClient({ baseUrl: BASE_URL, password: 's3cret' }); client.subscribeEvents({}); await until(() => connections.length === 1); expect(connections[0].url).toBe('/api/events?sessions=tui-no-terminal'); expect(connections[0].headers.authorization).toBe(`Basic ${Buffer.from('admin:s3cret').toString('base64')}`); expect(connections[0].headers.accept).toBe('text/event-stream'); }); it('subscribes to the terminal stream of named sessions when asked', async () => { client = new TuiClient({ baseUrl: BASE_URL }); client.subscribeEvents({}, { sessionIds: ['a', 'b'] }); await until(() => connections.length === 1); expect(connections[0].url).toBe('/api/events?sessions=a%2Cb'); }); it('reconnects when the stream ends', async () => { const statuses: Array<[string, TuiSseStatusDetail]> = []; client = new TuiClient({ baseUrl: BASE_URL }); const stream = client.subscribeEvents( { onStatus: (status, detail) => statuses.push([status, detail]) }, { baseBackoffMs: 10, maxBackoffMs: 20 } ); await until(() => stream.status === 'connected'); connections[0].res.end(); await until(() => connections.length === 2 && stream.status === 'connected'); expect(statuses.map(([status]) => status)).toEqual(['connected', 'reconnecting', 'connected']); expect(statuses[1][1].message).toBeTruthy(); }); it('reconnects when a live stream goes silent, which no socket error reports', async () => { client = new TuiClient({ baseUrl: BASE_URL }); client.subscribeEvents({}, { staleTimeoutMs: 150, checkIntervalMs: 25, baseBackoffMs: 10, maxBackoffMs: 20 }); await until(() => connections.length === 1); // The server holds the connection open and says nothing: exactly the case // the watchdog exists for. await until(() => connections.length === 2); expect(connections).toHaveLength(2); }); it('recommends polling once connecting keeps failing', async () => { refuse = true; const details: TuiSseStatusDetail[] = []; client = new TuiClient({ baseUrl: BASE_URL }); const stream = client.subscribeEvents( { onStatus: (_status, detail) => details.push(detail) }, { baseBackoffMs: 10, maxBackoffMs: 20, pollingAfterFailures: 2 } ); await until(() => details.length >= 2); expect(details[0]).toMatchObject({ attempt: 1, recommendPolling: false }); expect(details[1]).toMatchObject({ attempt: 2, recommendPolling: true }); expect(stream.recommendPolling).toBe(true); expect(stream.status).toBe('reconnecting'); }); it('stops reconnecting after close, so the process can exit', async () => { client = new TuiClient({ baseUrl: BASE_URL }); const stream = client.subscribeEvents({}, { baseBackoffMs: 10, maxBackoffMs: 20 }); await until(() => connections.length === 1); connections[0].res.end(); stream.close(); const seen = connections.length; await new Promise((resolve) => setTimeout(resolve, 120)); expect(connections.length).toBe(seen); }); it('closes every stream the client opened', async () => { client = new TuiClient({ baseUrl: BASE_URL }); client.subscribeEvents({}); client.subscribeEvents({}); await until(() => connections.length === 2); client.close(); await until(() => connections.every((connection) => connection.res.socket === null || connection.res.destroyed)); const seen = connections.length; await new Promise((resolve) => setTimeout(resolve, 120)); expect(connections.length).toBe(seen); }); it('refuses to subscribe before the client knows where the server is', () => { const disconnected = new TuiClient({ port: 3999 }); expect(() => disconnected.subscribeEvents({})).toThrow(/connect\(\)/); }); });