stack/packages/plugins/harness/pi/model-stream.mjs

import { firstOutputMarker } from '../../../harness/src/first-output.mjs';
import { resolve } from 'node:path';
import { fileURLToPath } from 'node:url';
import { spawn } from 'node:child_process';
import { constants } from 'node:os';
// The managed Pi adapter uses this monitor to translate its native JSON event
// streams back into the answer-only adapter contract. Startup, headers and
// keepalives are not model output; substantive deltas and named tool calls count.
export function modelEventStream({ runDir }) {
  const mark = firstOutputMarker(runDir);
  let pending = '', answer = null, failed = false;
  const line = value => {
    if (!value) return;
    const event = JSON.parse(value);
    const delta = event.type === 'message_update' && event.assistantMessageEvent;
    if (delta && ['text_delta', 'thinking_delta', 'toolcall_delta'].includes(delta.type) && typeof delta.delta === 'string' && delta.delta.length) mark(delta.delta);
    if (delta?.type === 'toolcall_start' && typeof delta.id === 'string' && delta.id.length && typeof delta.toolName === 'string' && delta.toolName.length) mark('output');
    if (event.type === 'agent_end') {
      const last = event.messages?.at(-1);
      if (last?.role === 'assistant') {
        failed = ['error', 'aborted'].includes(last.stopReason) || (last.stopReason === 'length' &&
          !(last.content ?? []).some(part => (part.type === 'text' && typeof part.text === 'string' && part.text.trim()) || part.type === 'toolCall'));
        answer = (last.content ?? []).filter(part => part.type === 'text').map(part => part.text + '\n').join('');
      }
    }
  };
  return {
    push(chunk) {
      pending += chunk;
      if (pending.length > 32 * 1024 * 1024) throw new Error('model event too large');
      let i; while ((i = pending.indexOf('\n')) !== -1) { line(pending.slice(0, i)); pending = pending.slice(i + 1); }
    },
    finish() {
      if (pending) line(pending);
      if (failed || typeof answer !== 'string') throw new Error('model stream did not complete');
      return answer;
    },
  };
}

// Invoked only by managed proxy adapter turns; the model CLI keeps its normal
// process group and stderr, while stdout remains the final answer alone.
if (process.argv[1] && resolve(process.argv[1]) === fileURLToPath(import.meta.url)) {
  const [, , harness, runDir, command, ...args] = process.argv;
  const stream = modelEventStream({ harness, runDir });
  const child = spawn(command, args, { stdio: ['ignore', 'pipe', 'inherit'] });
  let bad = false;
  child.stdout.setEncoding('utf8');
  child.stdout.on('data', chunk => {
    if (bad) return;
    try { stream.push(chunk); }
    catch { bad = true; child.kill('SIGTERM'); }
  });
  child.on('error', () => { bad = true; });
  child.on('close', (code, signal) => {
    if (bad) { process.stderr.write('model event stream failed\n'); process.exitCode = 1; return; }
    if (signal) {
      process.stderr.write(`model child killed by ${signal}\n`);
      const number = constants.signals[signal];
      process.exitCode = Number.isInteger(number) && number > 0 ? 128 + number : 1;
      return;
    }
    if (code !== 0) { process.exitCode = code ?? 1; return; }
    try { process.stdout.write(stream.finish()); }
    catch { process.stderr.write('model event stream failed\n'); process.exitCode = 1; }
  });
}