| 1 | import assert from 'node:assert/strict'; |
| 2 | import test from 'node:test'; |
| 3 | import { CodeWhaleRuntimeClient, isThreadStreamEnd } from '../index.js'; |
| 4 | |
| 5 | function clientFor(chunks, inspect = () => {}, headers = {}) { |
| 6 | return new CodeWhaleRuntimeClient({ token: 'fixture-token', fetch: async (url, init) => { |
| 7 | inspect(url, init); |
| 8 | return new Response(new ReadableStream({ start(controller) { |
| 9 | for (const chunk of chunks) controller.enqueue(new TextEncoder().encode(chunk)); |
| 10 | controller.close(); |
| 11 | } }), { headers: { 'content-type': 'text/event-stream; charset=utf-8', ...headers } }); |
| 12 | } }); |
| 13 | } |
| 14 | const collect = async stream => { const out = []; for await (const value of stream) out.push(value); return out; }; |
| 15 | |
| 16 | test('thread journal reads preserve the Runtime cursor, GET, cancellation and redirect boundary', async () => { |
| 17 | const signal = new AbortController().signal; |
| 18 | const record = { seq: 17, previous_seq: 9, event: 'item.completed', thread_id: 't/a', timestamp: '2026-09-12T00:00:00Z', payload: {} }; |
| 19 | const raw = `: keepalive\r\n\r\ndata: ${JSON.stringify(record)}\r\n\r\n`; |
| 20 | const client = clientFor([...raw], (url, init) => { |
| 21 | assert.equal(url.pathname, '/v1/threads/t%2Fa/events'); |
| 22 | assert.equal(url.searchParams.get('since_seq'), '9'); assert.equal(url.searchParams.get('replay_limit'), '100'); |
| 23 | assert.equal(init.method, 'GET'); assert.equal(init.redirect, 'error'); assert.equal(init.signal, signal); |
| 24 | assert.equal(init.headers.get('authorization'), 'Bearer fixture-token'); |
| 25 | }); |
| 26 | assert.deepEqual(await collect(client.threadEvents('t/a', { sinceSeq: 9, replayLimit: 100, signal })), [record]); |
| 27 | }); |
| 28 | test('thread stream refuses incomplete, oversized and invalid frames without emitting a partial event', async () => { |
| 29 | await assert.rejects(collect(clientFor(['data: {"seq":1}']).threadEvents('t')), /inside a frame/); |
| 30 | await assert.rejects(collect(clientFor(['data: ' + 'x'.repeat(2 * 1024 * 1024)]).threadEvents('t')), /size limit/); |
| 31 | await assert.rejects(collect(clientFor(['data: not-json\n\n']).threadEvents('t')), SyntaxError); |
| 32 | }); |
| 33 | test('thread stream rejects invalid cursor arguments and a JSON response', async () => { |
| 34 | for (const sinceSeq of [-1, 1.5, Number.MAX_SAFE_INTEGER + 1]) |
| 35 | await assert.rejects(collect(clientFor([]).threadEvents('t', { sinceSeq })), /safe integer/); |
| 36 | const client = new CodeWhaleRuntimeClient({ fetch: async () => new Response('{}', { headers: { 'content-type': 'application/json' } }) }); |
| 37 | await assert.rejects(collect(client.threadEvents('t')), /not an event stream/); |
| 38 | }); |
| 39 | |
| 40 | |
| 41 | test('thread progress is explicitly requested and remains separate from journal records at the same cursor', async () => { |
| 42 | const progress = { event: 'stream.progress', state: 'live', thread_id: 't', seq: 9 }; |
| 43 | const client = clientFor([`data: ${JSON.stringify(progress)}\n\n`], (url) => { |
| 44 | assert.equal(url.searchParams.get('progress'), 'true'); |
| 45 | }, { 'x-codewhale-event-progress': '1' }); |
| 46 | assert.deepEqual(await collect(client.threadEvents('t', { sinceSeq: 9, includeProgress: true })), [progress]); |
| 47 | await assert.rejects(collect(client.threadEvents('t', { includeProgress: 'yes' })), /boolean/); |
| 48 | }); |
| 49 | |
| 50 | test('thread progress fails explicitly and closes the stream when an older Runtime does not advertise it', async () => { |
| 51 | let canceled = false; |
| 52 | const client = new CodeWhaleRuntimeClient({ fetch: async () => new Response(new ReadableStream({ cancel() { canceled = true; } }), { headers: { 'content-type': 'text/event-stream' } }) }); |
| 53 | await assert.rejects(collect(client.threadEvents('t', { includeProgress: true })), error => error.capability === 'thread_event_progress' && error.status === 501); |
| 54 | assert.equal(canceled, true); |
| 55 | }); |
| 56 | |
| 57 | test('a server-ended thread stream yields its typed stream.end frame last', async () => { |
| 58 | const record = { seq: 17, previous_seq: 9, event: 'item.completed', thread_id: 't', timestamp: '2026-09-12T00:00:00Z', payload: {} }; |
| 59 | const end = { schema_version: 1, event: 'stream.end', kind: 'stream.end', thread_id: 't', reason: 'replay_failed', last_seq: 17, retryable: true }; |
| 60 | const raw = `event: item.completed\nid: 17\ndata: ${JSON.stringify(record)}\n\nevent: stream.end\ndata: ${JSON.stringify(end)}\n\n`; |
| 61 | const client = clientFor([raw], () => {}, { 'x-codewhale-stream-end': '1' }); |
| 62 | // The journal frame's SSE id becomes its cursor; stream.end carries none. |
| 63 | assert.deepEqual(await collect(client.threadEvents('t', { sinceSeq: 9 })), [{ ...record, cursor: '17' }, end]); |
| 64 | }); |
| 65 | |
| 66 | test('isThreadStreamEnd distinguishes end frames from journal and progress events', async () => { |
| 67 | const record = { seq: 17, event: 'future.journal.event', thread_id: 't', timestamp: '2026-09-12T00:00:00Z', payload: {} }; |
| 68 | const progress = { event: 'stream.progress', thread_id: 't', seq: 17, state: 'live' }; |
| 69 | const end = { schema_version: 1, event: 'stream.end', kind: 'stream.end', thread_id: 't', reason: 'runtime_shutdown', last_seq: 17, retryable: true }; |
| 70 | const raw = [record, progress, end].map(value => `data: ${JSON.stringify(value)}\n\n`).join(''); |
| 71 | const client = clientFor([raw], () => {}, { 'x-codewhale-event-progress': '1' }); |
| 72 | const events = await collect(client.threadEvents('t', { includeProgress: true })); |
| 73 | assert.deepEqual(events.map(isThreadStreamEnd), [false, false, true]); |
| 74 | assert.deepEqual(events.filter(isThreadStreamEnd), [end]); |
| 75 | }); |
| 76 |