import { describe, test, expect } from 'bun:test'; import { createTurnStream } from './turn-stream'; import type { ChatEvent, MessageCost } from '../../api/chat/types'; const SESSION = 'sess-1'; const COST: MessageCost = { inputTokens: 10, outputTokens: 20, totalUSD: 0.001 }; // Feed a whole event sequence through one stream and collect what came out, so the assertions read as // "this turn produced this transcript" rather than per-call plumbing. function run(events: ChatEvent[]) { const stream = createTurnStream(SESSION); const out = events.flatMap((e) => stream.push(e)); return { all: out, durable: out.filter((o) => o.durable).map((o) => o.msg), types: out.map((o) => o.msg.type), }; } describe('createTurnStream', () => { test('deltas are live-only and never durable', () => { const { all, durable } = run([ { type: 'delta', text: 'he' }, { type: 'delta', text: 'llo' }, ]); expect(all).toHaveLength(2); expect(all.every((o) => o.durable)).toBe(false); expect(durable).toHaveLength(0); }); test('the transcript id goes out live but is never persisted', () => { // Durable would put a second copy of the answer inside the question: the only way to replay this log // is to ask for it by the very id the event carries. const { all, durable, types } = run([{ type: 'session', claudeSessionId: 'claude-uuid-1' }]); expect(types).toEqual(['session:claude']); expect(all[0]!.msg).toEqual({ type: 'session:claude', claudeSessionId: 'claude-uuid-1' }); expect(durable).toHaveLength(0); }); test('the transcript id does not disturb an open delta buffer', () => { // It arrives at `system.init`, but a resumed turn can re-announce it mid-flight, and flushing there // would split one paragraph into two messages. const { durable } = run([ { type: 'delta', text: 'half ' }, { type: 'session', claudeSessionId: 'claude-uuid-1' }, { type: 'delta', text: 'a sentence' }, { type: 'result', cost: COST }, ]); expect(durable[0]).toEqual({ type: 'assistant:text', text: 'half a sentence' }); }); test('an explicit text event wins over the deltas that produced it', () => { const { durable } = run([ { type: 'delta', text: 'par' }, { type: 'delta', text: 'tial' }, { type: 'text', text: 'partial and complete' }, ]); expect(durable).toEqual([{ type: 'assistant:text', text: 'partial and complete' }]); }); test('a text event with no text falls back to the accumulated buffer', () => { const { durable } = run([ { type: 'delta', text: 'buffered' }, { type: 'text', text: '' }, ]); expect(durable).toEqual([{ type: 'assistant:text', text: 'buffered' }]); }); test('an empty text event with an empty buffer produces nothing', () => { expect(run([{ type: 'text', text: '' }]).all).toHaveLength(0); }); test('the buffer is flushed as one message before a tool call', () => { const { durable } = run([ { type: 'delta', text: 'Let me ' }, { type: 'delta', text: 'check.' }, { type: 'tool:start', toolCallId: 't1', toolName: 'Read', toolInput: { file: 'a.ts' } }, ]); expect(durable).toEqual([ { type: 'assistant:text', text: 'Let me check.' }, { type: 'tool:start', toolCallId: 't1', toolName: 'Read', toolInput: { file: 'a.ts' } }, ]); }); test('the buffer is flushed before the turn result, and result carries the session id', () => { const { durable } = run([ { type: 'delta', text: 'Done.' }, { type: 'result', cost: COST }, ]); expect(durable).toEqual([ { type: 'assistant:text', text: 'Done.' }, { type: 'result', sessionId: SESSION, cost: COST }, ]); }); test('a flushed buffer is not emitted twice', () => { const { durable } = run([ { type: 'delta', text: 'once' }, { type: 'tool:start', toolCallId: 't1', toolName: 'Read', toolInput: {} }, { type: 'tool:result', toolCallId: 't1', output: 'ok', isError: false }, { type: 'result', cost: COST }, ]); expect(durable.filter((m) => m.type === 'assistant:text')).toEqual([{ type: 'assistant:text', text: 'once' }]); }); test('deltas after a flush start a fresh block', () => { const { durable } = run([ { type: 'delta', text: 'first' }, { type: 'tool:start', toolCallId: 't1', toolName: 'Read', toolInput: {} }, { type: 'delta', text: 'second' }, { type: 'result', cost: COST }, ]); expect(durable.filter((m) => m.type === 'assistant:text')).toEqual([ { type: 'assistant:text', text: 'first' }, { type: 'assistant:text', text: 'second' }, ]); }); test('a full turn keeps the client-visible order', () => { const { types } = run([ { type: 'delta', text: 'a' }, { type: 'tool:start', toolCallId: 't1', toolName: 'Bash', toolInput: { cmd: 'ls' } }, { type: 'tool:result', toolCallId: 't1', output: 'a.ts', isError: false }, { type: 'delta', text: 'b' }, { type: 'result', cost: COST }, ]); expect(types).toEqual([ 'assistant:delta', 'assistant:text', 'tool:start', 'tool:result', 'assistant:delta', 'assistant:text', 'result', ]); }); test('errors and stops are durable, and do not flush a partial answer away', () => { const { durable } = run([ { type: 'delta', text: 'half' }, { type: 'error', message: 'boom' }, ]); expect(durable).toEqual([{ type: 'error', message: 'boom' }]); expect(run([{ type: 'stopped' }]).durable).toEqual([{ type: 'stopped' }]); }); test('background task events pass through and are durable — the reason the queue exists', () => { const { durable } = run([ { type: 'result', cost: COST }, { type: 'task:started', taskId: 'bg1', description: 'long job', taskType: 'local_agent' }, { type: 'task:notification', taskId: 'bg1', status: 'completed', summary: 'all good' }, ]); expect(durable).toEqual([ { type: 'result', sessionId: SESSION, cost: COST }, { type: 'task:started', taskId: 'bg1', description: 'long job', taskType: 'local_agent' }, { type: 'task:notification', taskId: 'bg1', status: 'completed', summary: 'all good' }, ]); }); test('compaction flushes the interrupted sentence before announcing itself', () => { const { durable } = run([ { type: 'delta', text: 'mid-thought' }, { type: 'compact:start', trigger: 'auto' }, { type: 'compact:done', trigger: 'auto', preTokens: 168215, durationMs: 154054 }, ]); expect(durable).toEqual([ { type: 'assistant:text', text: 'mid-thought' }, { type: 'compact:start', trigger: 'auto' }, { type: 'compact:done', trigger: 'auto', preTokens: 168215, durationMs: 154054 }, ]); }); test('streams are independent', () => { const a = createTurnStream('a'); const b = createTurnStream('b'); a.push({ type: 'delta', text: 'from-a' }); b.push({ type: 'delta', text: 'from-b' }); expect(a.push({ type: 'result', cost: COST })[0]!.msg).toEqual({ type: 'assistant:text', text: 'from-a' }); expect(b.push({ type: 'result', cost: COST })[0]!.msg).toEqual({ type: 'assistant:text', text: 'from-b' }); }); });