stack/packages/harness/tests/model-stream.test.mjs

import test from 'node:test';
import assert from 'node:assert/strict';
import fs from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import * as output from '../../plugins/harness/pi/model-stream.mjs';

for (const harness of ['pi']) {
  test(`${harness} counts a named tool call even when it has no input delta`, t => {
    const root = fs.mkdtempSync(join(tmpdir(), 'stream797-tool-'));
    t.after(() => fs.rmSync(root, { recursive: true, force: true }));
    const stream = output.modelEventStream({ harness, runDir: root });
    const event = harness === 'pi'
      ? {type:'message_update',assistantMessageEvent:{type:'toolcall_start',id:'tool_1',toolName:'read'}}
      : {type:'stream_event',event:{type:'content_block_start',content_block:{type:'tool_use',id:'tool_1',name:'read',input:{}}}};
    stream.push(JSON.stringify(event)+'\n');
    assert.equal(fs.readFileSync(join(root, 'proxy-first-output'), 'utf8'), 'output\n');
  });
  test(`${harness} marks real deltas, ignores startup events, and returns only the final answer`, t => {
    const root = fs.mkdtempSync(join(tmpdir(), 'stream797-'));
    t.after(() => fs.rmSync(root, { recursive: true, force: true }));
    assert.equal(typeof output.modelEventStream, 'function', 'shared event monitor exists');
    const stream = output.modelEventStream({ harness, runDir: root });
    const send = event => stream.push(JSON.stringify(event) + '\n');
    const file = join(root, 'proxy-first-output');
    send(harness === 'pi' ? {type:'message_update', assistantMessageEvent:{type:'start'}} : {type:'system',subtype:'init'});
    assert.equal(fs.existsSync(file), false);
    const delta = harness === 'pi' ? {type:'message_update', assistantMessageEvent:{type:'thinking_delta',delta:'secret thinking'}} : {type:'stream_event',event:{type:'content_block_delta',delta:{type:'thinking_delta',thinking:'secret thinking'}}};
    send(delta);
    assert.equal(fs.readFileSync(file, 'utf8'), 'output\n');
    send(harness === 'pi' ? {type:'agent_end',messages:[{role:'assistant',stopReason:'stop',content:[{type:'text',text:'final answer'}]}]} : {type:'result',subtype:'success',result:'final answer'});
    assert.equal(stream.finish(), 'final answer\n');
  });
  test(`${harness} does not mark empty deltas, pings, or startup events`, t => {
    const root = fs.mkdtempSync(join(tmpdir(), 'stream797-'));
    t.after(() => fs.rmSync(root, { recursive: true, force: true }));
    assert.equal(typeof output.modelEventStream, 'function', 'shared event monitor exists');
    const stream = output.modelEventStream({ harness, runDir: root });
    for (const e of [{type:'system'}, {type:'tool_result', content:'local tool output'}, {type:'message_end', message:{role:'toolResult', content:'local tool output'}}, {type:'stream_event',event:{type:'ping'}}, {type:'message_update',assistantMessageEvent:{type:'text_delta',delta:''}}, {type:'stream_event',event:{type:'content_block_delta',delta:{type:'text_delta',text:''}}}, {type:'stream_event',event:{type:'content_block_delta',delta:{type:'text_delta',text:{}}}}]) stream.push(JSON.stringify(e)+'\n');
    assert.equal(fs.existsSync(join(root, 'proxy-first-output')), false);
  });
}

test('monitor forwards only the answer and keeps model events private', async t => {
  const { spawn } = await import('node:child_process');
  const { fileURLToPath } = await import('node:url');
  const root = fs.mkdtempSync(join(tmpdir(), 'stream797-'));
  t.after(() => fs.rmSync(root, { recursive: true, force: true }));
  const monitor = fileURLToPath(new URL('../../plugins/harness/pi/model-stream.mjs', import.meta.url));
  const producer = `console.log(JSON.stringify({type:'message_update',assistantMessageEvent:{type:'text_delta',delta:'private'}}));console.log(JSON.stringify({type:'agent_end',messages:[{role:'assistant',stopReason:'stop',content:[{type:'text',text:'answer'}]}]}))`;
  const child = spawn(process.execPath, [monitor, 'pi', root, process.execPath, '-e', producer], { stdio:['ignore','pipe','pipe'] });
  let stdout='', stderr=''; child.stdout.on('data', c => stdout += c); child.stderr.on('data', c => stderr += c);
  const code = await new Promise(resolve => child.on('close', resolve));
  assert.equal(code, 0, stderr); assert.equal(stdout, 'answer\n');
  assert.equal(fs.readFileSync(join(root, 'proxy-first-output'), 'utf8'), 'output\n');
});

// Expected bytes come from the adapters' base text modes: Pi prints each text
// block with a newline; Claude terminates its result only if needed.
for (const [harness, parts, expected] of [
  ['pi', ['Plain answer.'], 'Plain answer.\n'],
  ['pi', ['Line one\n雪 👩🏽‍💻 café\nLine three'], 'Line one\n雪 👩🏽‍💻 café\nLine three\n'],
  ['pi', ['Line one\n雪 👩🏽‍💻 café\nLine three\n'], 'Line one\n雪 👩🏽‍💻 café\nLine three\n\n'],
  ['pi', [], ''], ['pi', [''], '\n'],
  ['pi', [null], ''], ['pi', ['First block.', 'Second block.\nUnicode 雪'], 'First block.\nSecond block.\nUnicode 雪\n'],
  ['claude', ['Plain answer.'], 'Plain answer.\n'],
  ['claude', ['Line one\n雪 👩🏽‍💻 café\nLine three'], 'Line one\n雪 👩🏽‍💻 café\nLine three\n'],
  ['claude', ['Line one\n雪 👩🏽‍💻 café\nLine three\n'], 'Line one\n雪 👩🏽‍💻 café\nLine three\n'],
  ['claude', [], '\n'], ['claude', [null], '\n'],
  ['claude', ['First block.', 'Second block.\nUnicode 雪'], 'Second block.\nUnicode 雪\n'],
]) {
  test(`${harness} preserves base stdout bytes for ${JSON.stringify(parts)}`, async t => {
    const { spawn } = await import('node:child_process');
    const { fileURLToPath } = await import('node:url');
    const root = fs.mkdtempSync(join(tmpdir(), 'stream797-parity-'));
    t.after(() => fs.rmSync(root, { recursive: true, force: true }));
    const monitor = fileURLToPath(new URL('../../plugins/harness/pi/model-stream.mjs', import.meta.url));
    const event = harness === 'pi'
      ? {type:'agent_end',messages:[{role:'assistant',stopReason:'stop',content:parts.map(text => text === null ? {type:'toolCall',id:'t',name:'read'} : {type:'text',text})}]}
      : {type:'result',subtype:'success',result:parts.filter(x => x !== null).at(-1) ?? ''};
    const helper = fileURLToPath(new URL('../../../adapters/claude/stream.mjs', import.meta.url));
    const args = harness === 'pi' ? [monitor, harness, root, process.execPath, '-e', `console.log(${JSON.stringify(JSON.stringify(event))})`] : [helper, process.execPath, '-e', `console.log(${JSON.stringify(JSON.stringify(event))})`];
    const child = spawn(process.execPath, args, {env:{...process.env, MOSAIC_MODEL_ACTIVITY_DIR:root}, stdio:['ignore','pipe','pipe','pipe','ignore','pipe']});
    child.stdio[3].resume(); child.stdio[5].resume();
    const chunks=[]; let stderr=''; child.stdout.on('data', c => chunks.push(c)); child.stderr.on('data', c => stderr += c);
    const code = await new Promise(resolve => child.on('close', resolve));
    assert.equal(code, 0, stderr); assert.deepEqual(Buffer.concat(chunks), Buffer.from(expected));
    assert.equal(fs.existsSync(join(root, 'proxy-first-output')), false);
  });
}

// Claude activity is owned by its actual adapter stream helper, not the Pi monitor.
for (const [name, events, marked] of [
  ['named tool call', [{type:'stream_event',event:{type:'content_block_start',content_block:{type:'tool_use',id:'tool_1',name:'read',input:{}}}}], true],
  ['real thinking delta', [{type:'system',subtype:'init'}, {type:'stream_event',event:{type:'content_block_delta',delta:{type:'thinking_delta',thinking:'private'}}}], true],
  ['local and synthetic events', [{type:'system',subtype:'init'}, {type:'system',subtype:'hook_response'}, {type:'user',message:{role:'user',content:[{type:'tool_result',content:'private'}]}}, {type:'assistant',error:'api_error',message:{role:'assistant',model:'<synthetic>',content:[{type:'text',text:'error'}]}}, {type:'result',subtype:'error',is_error:true,result:'error'}], false],
]) {
  test(`claude stream helper: ${name}`, async t => {
    const { spawn } = await import('node:child_process');
    const { fileURLToPath } = await import('node:url');
    const root = fs.mkdtempSync(join(tmpdir(), 'stream797-claude-'));
    t.after(() => fs.rmSync(root, {recursive:true,force:true}));
    const helper = fileURLToPath(new URL('../../../adapters/claude/stream.mjs', import.meta.url));
    const outputEvents = marked ? [...events, {type:'result',subtype:'success',result:'final answer'}] : events;
    const source = `for (const e of ${JSON.stringify(outputEvents)}) console.log(JSON.stringify(e))`;
    const child = spawn(process.execPath, [helper, process.execPath, '-e', source], {env:{...process.env,MOSAIC_MODEL_ACTIVITY_DIR:root},stdio:['ignore','pipe','pipe','pipe','ignore','pipe']});
    const chunks=[]; child.stdout.on('data', c => chunks.push(c)); child.stdio[3].resume(); child.stdio[5].resume();
    const code = await new Promise(resolve => child.on('close', resolve));
    assert.equal(code, marked ? 0 : 2);
    assert.equal(fs.existsSync(join(root,'proxy-first-output')), marked);
    if (marked) { assert.equal(fs.readFileSync(join(root,'proxy-first-output'),'utf8'), 'output\n'); assert.equal(Buffer.concat(chunks).toString(), 'final answer\n'); }
  });
}

test('Pi rejects an empty length stop, but accepts text or a tool call', t => {
  const root = fs.mkdtempSync(join(tmpdir(), 'length-'));
  t.after(() => fs.rmSync(root, { recursive: true, force: true }));
  for (const [content, fails] of [
    [[{type:'thinking', thinking:'private'}, {type:'text', text:'  '}], true],
    [[{type:'text', text:'answer'}], false],
    [[{type:'toolCall', id:'t', name:'read'}], false],
  ]) {
    const stream = output.modelEventStream({ harness:'pi', runDir:root });
    stream.push(JSON.stringify({type:'agent_end',messages:[{role:'assistant',stopReason:'length',content}]}) + '\n');
    if (fails) assert.throws(() => stream.finish(), /model stream did not complete/);
    else assert.doesNotThrow(() => stream.finish());
  }
});