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('the model the harness resolved for itself', () => { test('travels as an Officer id and is never persisted', () => { const { all, durable, types } = run([{ type: 'model:resolved', apiModel: 'claude-opus-5' }]); expect(types).toEqual(['model:resolved']); expect(all[0]!.msg).toEqual({ type: 'model:resolved', model: 'claude-code/opus' }); // Live-only for the same reason as `session:claude`: it marks no boundary, and a replay learns the // model from the session row instead. expect(durable).toHaveLength(0); }); test('an id with no honest Officer equivalent is dropped rather than forwarded raw', () => { // mythos has no catalogue entry on this machine, and `` is not a model at all. Offering // either to the picker is worse than leaving it as it was. expect(run([{ type: 'model:resolved', apiModel: 'claude-mythos-5' }]).all).toHaveLength(0); expect(run([{ type: 'model:resolved', apiModel: '' }]).all).toHaveLength(0); }); test('does not flush an open delta buffer', () => { // It says nothing about WHERE in the conversation anything happened, so cutting the sentence the // agent is mid-way through would be a boundary the event does not describe. const { types } = run([ { type: 'delta', text: 'thinking' }, { type: 'model:resolved', apiModel: 'claude-sonnet-5' }, ]); expect(types).toEqual(['assistant:delta', 'model:resolved']); }); }); 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 }, ]); }); // A tier switch has to survive a reload, and it has to land in the right PLACE. The marker says "opus // wrote everything above this line", which is false if it is emitted before the last thing opus said is // flushed out of the delta buffer. test('a model switch is durable and lands after the previous model’s last words', () => { const { durable } = run([ { type: 'delta', text: 'the old model speaking' }, { type: 'model:switched', from: 'claude-code/opus', to: 'claude-code/sonnet', ok: true }, { type: 'text', text: 'the new one' }, ]); expect(durable).toEqual([ { type: 'assistant:text', text: 'the old model speaking' }, { type: 'model:switched', from: 'claude-code/opus', to: 'claude-code/sonnet', ok: true }, { type: 'assistant:text', text: 'the new one' }, ]); }); /* Plan mode's three events, and the one property that is not obvious. A plan is the only DURABLE row the transcript needs in order to stay honest after a refresh: the card is interactive, so a plan that replayed without its decision would come back offering to approve something that was approved an hour ago. Both halves therefore have to survive, and `plan:proposed` has to land after the agent's introduction to it rather than inside the sentence. */ test('a plan is durable and lands after the words that introduce it', () => { const { durable } = run([ { type: 'delta', text: 'Here is what I propose:' }, { type: 'plan:proposed', requestId: 'plan_1', plan: '# Plan\n\nDo the thing.' }, ]); expect(durable).toEqual([ { type: 'assistant:text', text: 'Here is what I propose:' }, { type: 'plan:proposed', requestId: 'plan_1', plan: '# Plan\n\nDo the thing.' }, ]); }); test('a decision is durable and carries its feedback, so a rejected plan replays with its reason', () => { // The feedback IS what the agent was told to revise against, so the pair only makes sense together. const { durable } = run([{ type: 'plan:decided', requestId: 'plan_1', approved: false, feedback: 'too broad' }]); expect(durable).toEqual([{ type: 'plan:decided', requestId: 'plan_1', approved: false, feedback: 'too broad' }]); }); test('plan mode going on or off is a durable boundary, like a model switch', () => { // Same reasoning as the model marker: without it, a stretch where the agent researched and refused to // edit anything reads as an agent that lost its nerve. const { durable } = run([ { type: 'delta', text: 'ok' }, { type: 'plan:mode', enabled: true, ok: true }, ]); expect(durable).toEqual([ { type: 'assistant:text', text: 'ok' }, { type: 'plan:mode', enabled: true, ok: true }, ]); }); test('a refused plan-mode change is recorded with its reason', () => { const { durable } = run([{ type: 'plan:mode', enabled: true, ok: false, error: 'nope' }]); expect(durable).toEqual([{ type: 'plan:mode', enabled: true, ok: false, error: 'nope' }]); }); test('a refused switch is recorded too, with its reason', () => { // Silence here would be the worst outcome: the turn answers as the OLD model while the picker shows the // new one, which is the exact mismatch this whole change exists to remove. const { durable } = run([ { type: 'model:switched', from: 'claude-code/opus', to: 'claude-code/haiku', ok: false, error: 'nope' }, ]); expect(durable).toEqual([ { type: 'model:switched', from: 'claude-code/opus', to: 'claude-code/haiku', ok: false, error: 'nope' }, ]); }); 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' }); }); });