From 3aca21e5b9f4eef6958cfbcc8b2d50be6090ed0d Mon Sep 17 00:00:00 2001 From: Aaqibhafeez Khan Date: Sun, 25 Oct 2026 05:08:06 +0530 Subject: [PATCH 1/3] test: cover fragmented SSE framing and invalid UTF-8 --- tests/opencode-client.test.cjs | 58 ++++++++++++++++++++++++++++++++-- 1 file changed, 55 insertions(+), 3 deletions(-) diff --git a/tests/opencode-client.test.cjs b/tests/opencode-client.test.cjs index 00e8877..0282351 100644 --- a/tests/opencode-client.test.cjs +++ b/tests/opencode-client.test.cjs @@ -16,7 +16,7 @@ async function fixture(t, handler = () => false, options = {}) { if (await handler(req, res, item)) return; if (item.path === '/event') { res.writeHead(200, {'content-type': 'text/event-stream'}); streams.push(res); - res.write('event: message\r\ndata: {"type":"server.connected","properties":{}}\r\n\r\n'); return; + res.write('event: message\\r\\ndata: {"type":"server.connected","properties":{}}\\r\\n\\r\\n'); return; } res.writeHead(200, {'content-type': 'application/json'}); res.end(JSON.stringify(item.path === '/global/health' ? {healthy: true, version: '1.18.34'} : true)); @@ -31,7 +31,7 @@ test('pinned health, scoped authenticated HTTP, and actual SSE framing', async t const {client, seen, streams} = await fixture(t); await client.connect(); const received = once(client, 'event'); - const bytes = Buffer.from('data: {"type":"message.part.delta","properties":{"delta":"café"}}\n\n'); + const bytes = Buffer.from('data: {"type":"message.part.delta","properties":{"delta":"café"}}\\n\\n'); const split = bytes.indexOf(Buffer.from('é')) + 1; streams[0].write(bytes.subarray(0, split)); streams[0].write(bytes.subarray(split)); assert.equal((await received)[0].properties.delta, 'café'); @@ -73,7 +73,7 @@ test('JSON bounds, redirect rejection, prompt cancellation do not follow externa abort.abort(); await assert.rejects(pending, /cancelled/); }); test('invalid and oversized SSE close the client', async t => { - for (const frame of ['data: not-json\n\n', 'data: ' + 'x'.repeat(MAX_BODY + 1)]) { + for (const frame of ['data: not-json\\n\\n', 'data: ' + 'x'.repeat(MAX_BODY + 1)]) { const {client, streams} = await fixture(t); await client.connect(); const closed = once(client, 'closed'); streams[0].write(frame); await closed; assert.equal(client.closed, true); } @@ -86,3 +86,55 @@ test('session permission follows owner-supplied cooperative capability, not the assert.equal(permission?.action, cooperative ? 'allow' : undefined); } }); +test('SSE CRLF delimiters split across writes still dispatch the event', async t => { + const {client, streams} = await fixture(t); + await client.connect(); + const received = once(client, 'event'); + const frame = 'data: {"type":"message.part.delta","properties":{"delta":"split-crlf"}}'; + streams[0].write(frame + '\\r'); + streams[0].write('\\n\\r'); + streams[0].write('\\n'); + assert.equal((await received)[0].properties.delta, 'split-crlf'); +}); +test('multiple SSE events and comments in one write dispatch once in order', async t => { + const {client, streams} = await fixture(t); + await client.connect(); + const received = []; + const onEvent = event => { + received.push(event); + if (received.length === 2) client.removeListener('event', onEvent); + }; + client.on('event', onEvent); + streams[0].write([ + ': keepalive', + 'data: {"type":"message.part.delta","properties":{"delta":"first"}}', + '', + ': another comment', + 'data: {"type":"message.part.delta","properties":{"delta":"second"}}', + '', + ].join('\\n') + '\\n'); + await new Promise((resolve, reject) => { + const deadline = setTimeout(() => reject(new Error('timed out waiting for SSE events')), 500); + const check = () => { + if (received.length === 2) { clearTimeout(deadline); resolve(); } + else setImmediate(check); + }; + check(); + }); + assert.deepEqual(received.map(event => event.properties.delta), ['first', 'second']); +}); +test('malformed UTF-8 in an SSE data line closes the client without emitting an event', async t => { + const {client, streams} = await fixture(t); + await client.connect(); + let emitted = false; + client.on('event', () => { emitted = true; }); + const closed = once(client, 'closed'); + streams[0].write(Buffer.concat([ + Buffer.from('data: {"type":"message.part.delta","properties":{"delta":"'), + Buffer.from([0xc3, 0x28]), + Buffer.from('"}}\\n\\n'), + ])); + await closed; + assert.equal(client.closed, true); + assert.equal(emitted, false); +}); From 88d4eecadf7f16cc704d1cef0f6dc3354c65ef7f Mon Sep 17 00:00:00 2001 From: Aaqibhafeez Khan Date: Sun, 25 Oct 2026 05:08:17 +0530 Subject: [PATCH 2/3] test: correct SSE framing fixture escapes --- tests/opencode-client.test.cjs | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/tests/opencode-client.test.cjs b/tests/opencode-client.test.cjs index 0282351..a9defed 100644 --- a/tests/opencode-client.test.cjs +++ b/tests/opencode-client.test.cjs @@ -16,7 +16,7 @@ async function fixture(t, handler = () => false, options = {}) { if (await handler(req, res, item)) return; if (item.path === '/event') { res.writeHead(200, {'content-type': 'text/event-stream'}); streams.push(res); - res.write('event: message\\r\\ndata: {"type":"server.connected","properties":{}}\\r\\n\\r\\n'); return; + res.write('event: message\r\ndata: {"type":"server.connected","properties":{}}\r\n\r\n'); return; } res.writeHead(200, {'content-type': 'application/json'}); res.end(JSON.stringify(item.path === '/global/health' ? {healthy: true, version: '1.18.34'} : true)); @@ -31,7 +31,7 @@ test('pinned health, scoped authenticated HTTP, and actual SSE framing', async t const {client, seen, streams} = await fixture(t); await client.connect(); const received = once(client, 'event'); - const bytes = Buffer.from('data: {"type":"message.part.delta","properties":{"delta":"café"}}\\n\\n'); + const bytes = Buffer.from('data: {"type":"message.part.delta","properties":{"delta":"café"}}\n\n'); const split = bytes.indexOf(Buffer.from('é')) + 1; streams[0].write(bytes.subarray(0, split)); streams[0].write(bytes.subarray(split)); assert.equal((await received)[0].properties.delta, 'café'); @@ -73,7 +73,7 @@ test('JSON bounds, redirect rejection, prompt cancellation do not follow externa abort.abort(); await assert.rejects(pending, /cancelled/); }); test('invalid and oversized SSE close the client', async t => { - for (const frame of ['data: not-json\\n\\n', 'data: ' + 'x'.repeat(MAX_BODY + 1)]) { + for (const frame of ['data: not-json\n\n', 'data: ' + 'x'.repeat(MAX_BODY + 1)]) { const {client, streams} = await fixture(t); await client.connect(); const closed = once(client, 'closed'); streams[0].write(frame); await closed; assert.equal(client.closed, true); } @@ -91,9 +91,9 @@ test('SSE CRLF delimiters split across writes still dispatch the event', async t await client.connect(); const received = once(client, 'event'); const frame = 'data: {"type":"message.part.delta","properties":{"delta":"split-crlf"}}'; - streams[0].write(frame + '\\r'); - streams[0].write('\\n\\r'); - streams[0].write('\\n'); + streams[0].write(frame + '\r'); + streams[0].write('\n\r'); + streams[0].write('\n'); assert.equal((await received)[0].properties.delta, 'split-crlf'); }); test('multiple SSE events and comments in one write dispatch once in order', async t => { @@ -112,7 +112,7 @@ test('multiple SSE events and comments in one write dispatch once in order', asy ': another comment', 'data: {"type":"message.part.delta","properties":{"delta":"second"}}', '', - ].join('\\n') + '\\n'); + ].join('\n') + '\n'); await new Promise((resolve, reject) => { const deadline = setTimeout(() => reject(new Error('timed out waiting for SSE events')), 500); const check = () => { @@ -132,7 +132,7 @@ test('malformed UTF-8 in an SSE data line closes the client without emitting an streams[0].write(Buffer.concat([ Buffer.from('data: {"type":"message.part.delta","properties":{"delta":"'), Buffer.from([0xc3, 0x28]), - Buffer.from('"}}\\n\\n'), + Buffer.from('"}}\n\n'), ])); await closed; assert.equal(client.closed, true); From ac3d477adcab813f9fa8f94d6b7342c84d4bbd80 Mon Sep 17 00:00:00 2001 From: Aaqibhafeez Khan Date: Tue, 27 Oct 2026 05:56:29 +0530 Subject: [PATCH 3/3] test: harden SSE framing review cases --- tests/opencode-client.test.cjs | 71 ++++++++++++++++++++++------------ 1 file changed, 47 insertions(+), 24 deletions(-) diff --git a/tests/opencode-client.test.cjs b/tests/opencode-client.test.cjs index a9defed..37dfd42 100644 --- a/tests/opencode-client.test.cjs +++ b/tests/opencode-client.test.cjs @@ -22,7 +22,7 @@ async function fixture(t, handler = () => false, options = {}) { res.end(JSON.stringify(item.path === '/global/health' ? {healthy: true, version: '1.18.34'} : true)); }); server.listen(0, '127.0.0.1'); await once(server, 'listening'); - const client = new OpenCodeClient({baseUrl: `http://127.0.0.1:${server.address().port}`, password: PASSWORD, + const client = new OpenCodeClient({baseUrl: 'http://127.0.0.1:' + server.address().port, password: PASSWORD, timeoutMs: 1000, ...options}); t.after(() => { client.close(); for (const stream of streams) stream.destroy(); server.closeAllConnections(); server.close(); }); return {client, seen, streams}; @@ -38,7 +38,7 @@ test('pinned health, scoped authenticated HTTP, and actual SSE framing', async t await client.createSession({model: 'qwen3-0.6b-v1'}); await client.replyPermission('per_fixture', true); await client.replyPermission('per_other', false); assert.ok(seen.every(item => item.directory === '/workspace' && item.auth === - `Basic ${Buffer.from(`opencode:${PASSWORD}`).toString('base64')}`)); + 'Basic ' + Buffer.from('opencode:' + PASSWORD).toString('base64'))); assert.equal(seen[2].body.model.providerID, 'volparossa'); assert.deepEqual(seen.slice(3).map(item => item.body), [{reply: 'once'}, {reply: 'reject'}]); assert.ok(seen[2].body.permission.some(rule => rule.permission === '*' && rule.action === 'deny')); @@ -86,40 +86,63 @@ test('session permission follows owner-supplied cooperative capability, not the assert.equal(permission?.action, cooperative ? 'allow' : undefined); } }); -test('SSE CRLF delimiters split across writes still dispatch the event', async t => { +test('SSE CRLF delimiters split across observed response chunks still dispatch the event', async t => { const {client, streams} = await fixture(t); await client.connect(); const received = once(client, 'event'); const frame = 'data: {"type":"message.part.delta","properties":{"delta":"split-crlf"}}'; - streams[0].write(frame + '\r'); - streams[0].write('\n\r'); - streams[0].write('\n'); + const response = client.eventResponse; + const writeAndObserve = async segment => { + const observed = new Promise((resolve, reject) => { + let chunks = [], length = 0; + const onData = chunk => { + chunks.push(Buffer.from(chunk)); + length += chunk.length; + if (length >= Buffer.byteLength(segment)) { + response.off('data', onData); + resolve(Buffer.concat(chunks)); + } + }; + response.on('data', onData); + response.once('error', reject); + }); + streams[0].write(segment); + assert.equal((await observed).toString(), segment); + }; + await writeAndObserve(frame + '\r'); + await writeAndObserve('\n\r'); + await writeAndObserve('\n'); assert.equal((await received)[0].properties.delta, 'split-crlf'); }); test('multiple SSE events and comments in one write dispatch once in order', async t => { const {client, streams} = await fixture(t); await client.connect(); const received = []; - const onEvent = event => { - received.push(event); - if (received.length === 2) client.removeListener('event', onEvent); - }; - client.on('event', onEvent); - streams[0].write([ - ': keepalive', - 'data: {"type":"message.part.delta","properties":{"delta":"first"}}', - '', - ': another comment', - 'data: {"type":"message.part.delta","properties":{"delta":"second"}}', - '', - ].join('\n') + '\n'); await new Promise((resolve, reject) => { - const deadline = setTimeout(() => reject(new Error('timed out waiting for SSE events')), 500); - const check = () => { - if (received.length === 2) { clearTimeout(deadline); resolve(); } - else setImmediate(check); + let quietTimer; + const deadlineTimer = setTimeout(() => finish(new Error('timed out waiting for SSE events')), 500); + const finish = error => { + clearTimeout(deadlineTimer); + clearTimeout(quietTimer); + client.off('event', onEvent); + error ? reject(error) : resolve(); + }; + const onEvent = event => { + received.push(event); + if (received.length >= 2) { + clearTimeout(quietTimer); + quietTimer = setTimeout(() => finish(), 50); + } }; - check(); + client.on('event', onEvent); + streams[0].write([ + ': keepalive', + 'data: {"type":"message.part.delta","properties":{"delta":"first"}}', + '', + ': another comment', + 'data: {"type":"message.part.delta","properties":{"delta":"second"}}', + '', + ].join('\n') + '\n'); }); assert.deepEqual(received.map(event => event.properties.delta), ['first', 'second']); });