82 lines
3.9 KiB
JavaScript
82 lines
3.9 KiB
JavaScript
// Consume Claude Code's SDK event stream privately. A complete final result
|
|
// reaches stdout, including errors, with the same newline as text mode.
|
|
import { spawn } from 'node:child_process';
|
|
import { writeSync } from 'node:fs';
|
|
import { constants } from 'node:os';
|
|
import { firstOutputMarker } from '../../packages/harness/src/first-output.mjs';
|
|
|
|
const mark = process.env.MOSAIC_MODEL_ACTIVITY_DIR
|
|
? firstOutputMarker(process.env.MOSAIC_MODEL_ACTIVITY_DIR) : () => {};
|
|
const [binary, ...args] = process.argv.slice(2);
|
|
const child = spawn(binary, args, { stdio: ['ignore', 'pipe', 5] });
|
|
let pending = '', answer, resultSeen = false, resultFailed = false, failed = false, killTimer;
|
|
const fail = () => {
|
|
if (failed) return;
|
|
failed = true;
|
|
child.kill('SIGTERM');
|
|
killTimer = setTimeout(() => child.kill('SIGKILL'), 5000).unref();
|
|
};
|
|
function event(line) {
|
|
if (!line.trim()) return;
|
|
const item = JSON.parse(line);
|
|
const model = item.type === 'stream_event' ? item.event : null;
|
|
// Init/status, hooks, user/tool results and arbitrary local writes aren't
|
|
// model activity. message_start is the upstream assistant response header;
|
|
// partial blocks also cover models that first respond with thinking/tools.
|
|
const start = model?.type === 'message_start' && model.message?.role === 'assistant';
|
|
const block = model?.type === 'content_block_start'
|
|
&& ['text', 'thinking', 'tool_use'].includes(model.content_block?.type);
|
|
const delta = model?.type === 'content_block_delta'
|
|
&& ['text_delta', 'thinking_delta', 'input_json_delta'].includes(model.delta?.type);
|
|
const assistant = item.type === 'assistant' && !item.error
|
|
&& item.message?.model !== '<synthetic>' && item.message?.role === 'assistant'
|
|
&& item.message.content?.some(c => ['text', 'thinking', 'tool_use'].includes(c.type));
|
|
if (start || block || delta || assistant) mark('activity');
|
|
if (item.type === 'result') {
|
|
if (resultSeen) throw Error();
|
|
resultSeen = true;
|
|
// Let a normal CLI failure exit on its own, retaining its exit code.
|
|
resultFailed = item.subtype !== 'success' || !!item.is_error;
|
|
if (typeof item.result === 'string') answer = item.result;
|
|
else if (!resultFailed) throw Error();
|
|
}
|
|
}
|
|
child.stdout.setEncoding('utf8');
|
|
child.stdout.on('data', chunk => {
|
|
if (failed) return;
|
|
try {
|
|
pending += chunk;
|
|
let end;
|
|
while ((end = pending.indexOf('\n')) !== -1) {
|
|
if (end > 16 * 1024 * 1024) throw Error();
|
|
event(pending.slice(0, end));
|
|
pending = pending.slice(end + 1);
|
|
}
|
|
if (pending.length > 16 * 1024 * 1024) throw Error();
|
|
} catch { fail(); }
|
|
});
|
|
child.on('error', fail);
|
|
child.on('close', (code, signal) => {
|
|
clearTimeout(killTimer);
|
|
if (!failed && pending) {
|
|
try { event(pending); } catch { fail(); }
|
|
}
|
|
// Integrity and rendering are separate from the CLI's success classification.
|
|
// Buffer until close so corrupt/truncated streams never leak a final result.
|
|
if (!failed && (!resultSeen || signal)) fail();
|
|
if (signal) writeSync(5, `model child killed by ${signal}\n`);
|
|
const number = signal && constants.signals[signal];
|
|
const signalStatus = signal ? (Number.isInteger(number) && number > 0 ? 128 + number : 1) : 0;
|
|
const status = process.exitCode = signalStatus || code || (failed || resultFailed ? 2 : 0);
|
|
// Only a fully consumed stream whose answer has been handed to stdout can
|
|
// acknowledge completion, and the token names the status it will exit with.
|
|
// A helper that dies after acknowledging exits with a different status, so
|
|
// the adapter still reports it. The adapter owns the fixed diagnostic,
|
|
// including when this helper never loads or dies early. spawn's explicit
|
|
// stdio keeps fd 3 out of the CLI and its descendants.
|
|
const acknowledge = error => writeSync(3, failed || error ? 'invalid' : `complete ${status}`);
|
|
if (!failed && answer !== undefined) {
|
|
process.stdout.write(answer.endsWith('\n') ? answer : answer + '\n', acknowledge);
|
|
} else acknowledge();
|
|
});
|