jason.woltje
3c689efa31
Pi could spend its output limit on thinking and the runner relayed an empty answer as a result. Preserve final stop metadata and allow two bounded continuations, accounting for the whole message in one transcript usage delta. Record exhaustion on the message trail and notify the chief without an automatic reply cycle. Mount the shared notice prefix in the sandbox.
295 lines
17 KiB
JavaScript
295 lines
17 KiB
JavaScript
// Real adapter and watchdog, with a fake CLI: no models, proxy or credentials.
|
|
import test from 'node:test';
|
|
import assert from 'node:assert/strict';
|
|
import { spawn, spawnSync } from 'node:child_process';
|
|
import { existsSync, mkdirSync, readFileSync, statSync, symlinkSync, unlinkSync, writeFileSync } from 'node:fs';
|
|
import { dirname, join } from 'node:path';
|
|
import { watchModelOutput } from '../../cli/src/model-stall.mjs';
|
|
import { firstTurnMarker } from '../src/first-output.mjs';
|
|
import { adapterEnv } from '../src/runner.mjs';
|
|
import piExtension from '../src/pi-extension.mjs';
|
|
import { REPO, scratch } from './helpers.mjs';
|
|
|
|
const ADAPTER = join(REPO, 'adapters/claude/adapter.sh');
|
|
const FAKE = new URL('./fixtures/claude-activity.mjs', import.meta.url).pathname;
|
|
const delay = ms => new Promise(resolve => setTimeout(resolve, ms));
|
|
function fixture(t, mode = 'tool') {
|
|
const runDir = scratch(t), workspace = join(runDir, 'ws');
|
|
mkdirSync(workspace);
|
|
for (const name of ['prompt', 'settings', 'mcp']) writeFileSync(join(runDir, name), '{}');
|
|
const cli = join(runDir, 'claude');
|
|
writeFileSync(cli, `#!/bin/sh\nexec '${process.execPath}' '${FAKE}' "$@"\n`, { mode: 0o700 });
|
|
const session = {
|
|
runDir, workspace, sessionDir: join(runDir, 'session'), model: 'proxy-model',
|
|
bundle: { prompt: join(runDir, 'prompt') },
|
|
harnessEnv: { MOSAIC_CLAUDE_SETTINGS: join(runDir, 'settings'), MOSAIC_CLAUDE_MCP_CONFIG: join(runDir, 'mcp'), MOSAIC_THINKING: 'high' },
|
|
};
|
|
const env = adapterEnv(session, {
|
|
PATH: `${dirname(process.execPath)}:/usr/bin:/bin`, MOSAIC_CLAUDE_BIN: cli,
|
|
MOSAIC_REQUEST: 'write a file', FAKE_CLAUDE_MODE: mode,
|
|
});
|
|
return { runDir, workspace, env, cli };
|
|
}
|
|
function start(t, f) {
|
|
const child = spawn('/bin/sh', [f.adapter ?? ADAPTER], { env: f.env, stdio: ['ignore', 'pipe', 'pipe'], detached: true });
|
|
const kill = () => { try { process.kill(-child.pid, 'SIGTERM'); } catch {} };
|
|
let stdout = Buffer.alloc(0), stderr = '';
|
|
child.stdout.on('data', b => { stdout = Buffer.concat([stdout, b]); });
|
|
child.stderr.on('data', b => { stderr += b; });
|
|
const done = new Promise((resolve, reject) => {
|
|
child.once('error', reject);
|
|
child.once('close', (code, signal) => resolve({ code, signal, stdout, stderr }));
|
|
});
|
|
t.after(async () => { kill(); await done; });
|
|
return { child, done, kill, stdout: () => stdout };
|
|
}
|
|
|
|
for (const mode of ['tool', 'thinking', 'header', 'assistant']) {
|
|
test(`Claude ${mode} activity survives a short watchdog window before final stdout`, { timeout: 10_000 }, async t => {
|
|
const f = fixture(t, mode);
|
|
f.env.FAKE_CLAUDE_DELAY = '1400';
|
|
firstTurnMarker(f.runDir)();
|
|
const turn = start(t, f), receipts = [];
|
|
let stops = 0;
|
|
const watch = watchModelOutput({ runDir: f.runDir, seconds: 0.8, armAtFirstTurn: true, keyId: '0123456789ab',
|
|
stop: async () => { stops++; turn.kill(); }, receipt: r => receipts.push(r) });
|
|
t.after(watch.cancel);
|
|
await watch.done;
|
|
assert.equal(stops, 0);
|
|
assert.equal(turn.stdout().length, 0, 'model events stay private while the tool turn is in progress');
|
|
assert.equal(readFileSync(join(f.runDir, 'proxy-first-output'), 'utf8'), 'output\n');
|
|
assert.equal(statSync(join(f.runDir, 'proxy-first-output')).mode & 0o777, 0o600);
|
|
const result = await turn.done;
|
|
assert.equal(result.code, 0, result.stderr);
|
|
// Compare exact bytes against the same CLI's original text format.
|
|
const text = spawnSync(f.cli, ['--output-format', 'text'], { env: f.env });
|
|
assert.equal(text.status, 0);
|
|
assert.deepEqual(result.stdout, text.stdout);
|
|
assert.deepEqual(receipts, []);
|
|
assert.match(readFileSync(join(f.runDir, 'session/claude-session-id'), 'utf8'), /^[0-9a-f-]{36}\n$/);
|
|
});
|
|
}
|
|
|
|
test('silent Claude is stopped with proxy-stall despite init, tool results, stderr and workspace writes', { timeout: 10_000 }, async t => {
|
|
const f = fixture(t, 'silent'), turn = start(t, f), receipts = [];
|
|
firstTurnMarker(f.runDir)();
|
|
const watch = watchModelOutput({ runDir: f.runDir, seconds: 0.8, armAtFirstTurn: true, keyId: '0123456789ab',
|
|
stop: async () => turn.kill(), receipt: r => receipts.push(r) });
|
|
t.after(watch.cancel);
|
|
await watch.done;
|
|
const result = await turn.done;
|
|
assert.notEqual(result.code, 0);
|
|
assert.equal(result.stdout.length, 0);
|
|
assert.equal(existsSync(join(f.workspace, 'draft')), true, 'workspace activity alone is insufficient');
|
|
assert.equal(existsSync(join(f.runDir, 'proxy-first-output')), false);
|
|
assert.equal(existsSync(join(f.runDir, 'session/claude-session-id')), false);
|
|
assert.deepEqual(receipts, [{ action: 'proxy-stall', keyIds: ['0123456789ab'] }]);
|
|
});
|
|
|
|
for (const resumed of [false, true]) {
|
|
test(`interrupted Claude activity emits no partial answer, preserves session state and kills its CLI (resume=${resumed})`, { timeout: 10_000 }, async t => {
|
|
const f = fixture(t, 'interrupt');
|
|
const sessionFile = join(f.runDir, 'session/claude-session-id');
|
|
if (resumed) {
|
|
mkdirSync(join(f.runDir, 'session'));
|
|
writeFileSync(sessionFile, 'previous-session-id\n');
|
|
}
|
|
const turn = start(t, f);
|
|
const deadline = Date.now() + 5000;
|
|
while (!existsSync(join(f.runDir, 'proxy-first-output'))) {
|
|
assert.ok(Date.now() < deadline, 'activity marker appears');
|
|
await delay(10);
|
|
}
|
|
const pid = Number(readFileSync(join(f.workspace, 'cli-pid'), 'utf8'));
|
|
turn.kill();
|
|
const result = await turn.done;
|
|
assert.notEqual(result.code, 0);
|
|
assert.equal(result.stdout.length, 0);
|
|
if (resumed) assert.equal(readFileSync(sessionFile, 'utf8'), 'previous-session-id\n');
|
|
else assert.equal(existsSync(sessionFile), false);
|
|
// The CLI is a grandchild; its pipe can close just before the OS reaps it.
|
|
const reapDeadline = Date.now() + 5000;
|
|
for (;;) {
|
|
try { process.kill(pid, 0); }
|
|
catch (error) { assert.equal(error.code, 'ESRCH'); break; }
|
|
assert.ok(Date.now() < reapDeadline, 'interrupted CLI is reaped');
|
|
await delay(10);
|
|
}
|
|
});
|
|
}
|
|
|
|
for (const mode of ['malformed', 'missing', 'malformed-after-result', 'oversized', 'truncated']) {
|
|
test(`Claude ${mode} stream fails closed without raw event diagnostics or a session receipt`, async t => {
|
|
const f = fixture(t, mode), result = await start(t, f).done;
|
|
assert.notEqual(result.code, 0);
|
|
assert.equal(result.stdout.length, 0);
|
|
assert.match(result.stderr, /claude adapter: invalid or incomplete model stream/);
|
|
assert.doesNotMatch(result.stderr, /credential-canary/);
|
|
assert.equal(existsSync(join(f.runDir, 'session/claude-session-id')), false);
|
|
});
|
|
}
|
|
|
|
test('Claude helper marker failure keeps the fixed diagnostic and emits no result', async t => {
|
|
const f = fixture(t);
|
|
// A file where the marker directory should be makes the private helper fail.
|
|
f.env.MOSAIC_MODEL_ACTIVITY_DIR = join(f.runDir, 'prompt');
|
|
const result = await start(t, f).done;
|
|
assert.notEqual(result.code, 0);
|
|
assert.equal(result.stdout.length, 0);
|
|
assert.match(result.stderr, /claude adapter: invalid or incomplete model stream/);
|
|
assert.doesNotMatch(result.stderr, /credential-canary|ENOTDIR/);
|
|
assert.equal(existsSync(join(f.runDir, 'session/claude-session-id')), false);
|
|
assert.equal(existsSync(join(f.runDir, 'proxy-first-output')), false);
|
|
});
|
|
|
|
// These failures happen outside the helper's internal fail() handler.
|
|
for (const mode of ['sigkill', 'missing', 'static-import', 'runtime']) {
|
|
test(`Claude helper ${mode} failure reports only the fixed diagnostic without activity or a session`, { timeout: 10_000 }, async t => {
|
|
const f = fixture(t);
|
|
if (mode === 'sigkill') {
|
|
writeFileSync(f.cli, `#!/bin/sh\nexec '${process.execPath}' -e 'process.kill(process.ppid, "SIGKILL")'\n`);
|
|
} else {
|
|
f.adapter = join(f.runDir, 'adapter.sh');
|
|
symlinkSync(ADAPTER, f.adapter);
|
|
// The adapter resolves the helper beside its invocation path. Do not
|
|
// modify repository source to inject a missing/broken module.
|
|
if (mode !== 'missing') writeFileSync(join(f.runDir, 'stream.mjs'), mode === 'static-import'
|
|
? "import './credential-canary-missing.mjs';\n"
|
|
: "throw Error('credential-canary-runtime');\n");
|
|
}
|
|
const result = await start(t, f).done;
|
|
assert.notEqual(result.code, 0);
|
|
assert.equal(result.stdout.length, 0);
|
|
assert.equal(result.stderr, 'claude adapter: invalid or incomplete model stream\n');
|
|
assert.equal(existsSync(join(f.runDir, 'session/claude-session-id')), false);
|
|
assert.equal(existsSync(join(f.runDir, 'proxy-first-output')), false);
|
|
});
|
|
}
|
|
|
|
// The helper can die after its result is complete but while a large answer is
|
|
// still draining. That must not look like a complete CLI result.
|
|
test('Claude helper killed while its answer drains reports the fixed diagnostic', { timeout: 20_000 }, async t => {
|
|
const f = fixture(t);
|
|
const script = join(f.runDir, 'cli.cjs');
|
|
writeFileSync(script, `require('node:fs').writeFileSync(${JSON.stringify(join(f.workspace, 'helper-pid'))}, String(process.ppid));
|
|
process.stdout.write(JSON.stringify({ type: 'result', subtype: 'success', result: 'x'.repeat(8 << 20) }) + '\\n');`);
|
|
writeFileSync(f.cli, `#!/bin/sh\nexec '${process.execPath}' '${script}' "$@"\n`);
|
|
const child = spawn('/bin/sh', [ADAPTER], { env: f.env, stdio: ['ignore', 'pipe', 'pipe'], detached: true });
|
|
t.after(() => { try { process.kill(-child.pid, 'SIGKILL'); } catch {} });
|
|
let stderr = '', bytes = 0;
|
|
child.stderr.on('data', b => { stderr += b; });
|
|
const done = new Promise(resolve => child.once('close', code => resolve(code)));
|
|
// Leave stdout unread so the answer cannot drain, then kill the helper.
|
|
await new Promise(resolve => child.stdout.once('readable', resolve));
|
|
await delay(150);
|
|
process.kill(Number(readFileSync(join(f.workspace, 'helper-pid'), 'utf8')), 'SIGKILL');
|
|
child.stdout.on('data', b => { bytes += b.length; });
|
|
const code = await done;
|
|
assert.equal(code, 137);
|
|
assert.ok(bytes < (8 << 20) + 1, 'the answer was cut short');
|
|
assert.equal(stderr, 'claude adapter: invalid or incomplete model stream\n');
|
|
assert.equal(existsSync(join(f.runDir, 'session/claude-session-id')), false);
|
|
});
|
|
|
|
// A handled CLI result can return the same codes as a dead/unloaded helper.
|
|
for (const code of [1, 137]) {
|
|
test(`Claude complete CLI error exit ${code} preserves text bytes without a helper diagnostic`, async t => {
|
|
const f = fixture(t);
|
|
const script = join(f.runDir, 'cli.mjs');
|
|
writeFileSync(script, `const answer = 'API Error: fixture é 🦆\\n\\n';
|
|
process.stdout.write(process.argv.includes('text') ? answer
|
|
: JSON.stringify({ type: 'result', subtype: 'success', is_error: true, result: answer }) + '\\n');
|
|
process.exitCode = ${code};`);
|
|
writeFileSync(f.cli, `#!/bin/sh\nexec '${process.execPath}' '${script}' "$@"\n`);
|
|
const result = await start(t, f).done;
|
|
const text = spawnSync(f.cli, ['--output-format', 'text'], { env: f.env });
|
|
assert.equal(text.status, code);
|
|
assert.equal(result.code, text.status);
|
|
assert.deepEqual(result.stdout, text.stdout);
|
|
assert.equal(result.stderr, '');
|
|
assert.equal(existsSync(join(f.runDir, 'session/claude-session-id')), false);
|
|
assert.equal(existsSync(join(f.runDir, 'proxy-first-output')), false);
|
|
});
|
|
}
|
|
|
|
for (const mode of ['failed', 'failed-success-subtype', 'failed-zero-exit', 'nonzero-success']) {
|
|
test(`Claude ${mode} renders the final result independently of failure without a session or synthetic activity`, async t => {
|
|
const f = fixture(t, mode), result = await start(t, f).done;
|
|
assert.equal(result.code, mode === 'failed-zero-exit' ? 2 : 7);
|
|
const answer = spawnSync(f.cli, ['--output-format', 'text'], { env: f.env });
|
|
assert.deepEqual(result.stdout, answer.stdout);
|
|
assert.doesNotMatch(result.stderr, /invalid or incomplete model stream/);
|
|
assert.equal(existsSync(join(f.runDir, 'session/claude-session-id')), false);
|
|
assert.equal(existsSync(join(f.runDir, 'proxy-first-output')), false);
|
|
});
|
|
}
|
|
|
|
test('Pi extension marks real streamed thinking/tools/text, never local activity or user/tool messages', async t => {
|
|
const runDir = scratch(t);
|
|
writeFileSync(join(runDir, 'policy'), JSON.stringify({ harness: 'pi', tools: [], typed: [] }));
|
|
writeFileSync(join(runDir, 'tools'), '[]');
|
|
const values = { MOSAIC_POLICY_FILE: join(runDir, 'policy'), MOSAIC_TOOLS_FILE: join(runDir, 'tools'),
|
|
MOSAIC_TOOL_SOCKET: join(runDir, 'sock'), MOSAIC_TURN_MARKER: join(runDir, 'done'), MOSAIC_MODEL_ACTIVITY_DIR: runDir };
|
|
const prior = Object.fromEntries(Object.keys(values).map(k => [k, process.env[k]]));
|
|
Object.assign(process.env, values);
|
|
t.after(() => { for (const [k, v] of Object.entries(prior)) { if (v === undefined) delete process.env[k]; else process.env[k] = v; } });
|
|
for (const type of ['thinking_delta', 'toolcall_start', 'toolcall_delta', 'text_delta']) {
|
|
const handlers = new Map();
|
|
piExtension({ on: (name, fn) => handlers.set(name, fn), registerTool() {} });
|
|
const update = handlers.get('message_update');
|
|
for (const role of ['user', 'toolResult']) await update({ message: { role }, assistantMessageEvent: { type } });
|
|
await update({ message: { role: 'assistant' }, assistantMessageEvent: { type: 'start' } });
|
|
assert.equal(existsSync(join(runDir, 'proxy-first-output')), false);
|
|
await update({ message: { role: 'assistant' }, assistantMessageEvent: { type } });
|
|
assert.equal(readFileSync(join(runDir, 'proxy-first-output'), 'utf8'), 'output\n');
|
|
// A new producer on each case proves each event type on its own.
|
|
unlinkSync(join(runDir, 'proxy-first-output'));
|
|
}
|
|
});
|
|
|
|
test('managed Pi uses the strict monitor, not extension-local activity', async t => {
|
|
const runDir = scratch(t);
|
|
writeFileSync(join(runDir, 'policy'), JSON.stringify({harness:'pi', tools:[], typed:[]}));
|
|
writeFileSync(join(runDir, 'tools'), '[]');
|
|
const values = { MOSAIC_POLICY_FILE:join(runDir,'policy'), MOSAIC_TOOLS_FILE:join(runDir,'tools'), MOSAIC_TOOL_SOCKET:join(runDir,'sock'), MOSAIC_TURN_MARKER:join(runDir,'done'), MOSAIC_MODEL_ACTIVITY_DIR:runDir, MOSAIC_MODEL_EVENT_MONITOR:'managed' };
|
|
const prior = Object.fromEntries(Object.keys(values).map(k => [k, process.env[k]]));
|
|
Object.assign(process.env, values);
|
|
t.after(() => { for (const [k,v] of Object.entries(prior)) { if (v === undefined) delete process.env[k]; else process.env[k] = v; } });
|
|
const handlers = new Map();
|
|
piExtension({on:(name,fn) => handlers.set(name,fn), registerTool(){}});
|
|
for (const type of ['text_delta','thinking_delta','toolcall_start','toolcall_delta'])
|
|
await handlers.get('message_update')({message:{role:'assistant'},assistantMessageEvent:{type}});
|
|
assert.equal(existsSync(join(runDir,'proxy-first-output')),false);
|
|
});
|
|
|
|
test('Claude adapter preserves text-mode bytes for a tool-only final turn', async t => {
|
|
const f = fixture(t);
|
|
const script = join(f.runDir, 'tool-only.cjs');
|
|
writeFileSync(script, `if (process.argv.includes('text')) process.stdout.write('\\n');
|
|
else { console.log(JSON.stringify({type:'stream_event',event:{type:'content_block_start',content_block:{type:'tool_use',id:'tool_1',name:'read',input:{}}}}));
|
|
console.log(JSON.stringify({type:'result',subtype:'success',result:''})); }`);
|
|
writeFileSync(f.cli, `#!/bin/sh\nexec '${process.execPath}' '${script}' "$@"\n`);
|
|
const text = spawnSync(f.cli, ['--output-format','text'], {env:f.env});
|
|
const streamed = await start(t, f).done;
|
|
assert.equal(text.status,0); assert.equal(streamed.code,0,streamed.stderr);
|
|
assert.deepEqual(streamed.stdout,text.stdout); assert.deepEqual(streamed.stdout,Buffer.from('\n'));
|
|
assert.equal(readFileSync(join(f.runDir,'proxy-first-output'),'utf8'),'output\n');
|
|
});
|
|
|
|
test('Pi turn marker records final assistant stop reason and output presence', async t => {
|
|
const runDir = scratch(t);
|
|
writeFileSync(join(runDir, 'policy'), JSON.stringify({harness:'pi', tools:[], typed:[]}));
|
|
writeFileSync(join(runDir, 'tools'), '[]');
|
|
const values = { MOSAIC_POLICY_FILE:join(runDir,'policy'), MOSAIC_TOOLS_FILE:join(runDir,'tools'), MOSAIC_TOOL_SOCKET:join(runDir,'sock'), MOSAIC_TURN_MARKER:join(runDir,'done') };
|
|
const prior = Object.fromEntries(Object.keys(values).map(k => [k, process.env[k]]));
|
|
Object.assign(process.env, values);
|
|
t.after(() => { for (const [k,v] of Object.entries(prior)) { if (v === undefined) delete process.env[k]; else process.env[k] = v; } });
|
|
const handlers = new Map();
|
|
piExtension({on:(name,fn) => handlers.set(name,fn), registerTool(){}});
|
|
for (const [content, hasText, hasToolCall] of [[[{type:'thinking',thinking:'x'}], false, false], [[{type:'text',text:'answer'}], true, false], [[{type:'toolCall',name:'read'}], false, true]]) {
|
|
await handlers.get('agent_end')({messages:[{role:'assistant',stopReason:'length',content}]});
|
|
const marker = JSON.parse(readFileSync(join(runDir,'done'),'utf8'));
|
|
assert.equal(marker.stopReason, 'length'); assert.equal(marker.hasText, hasText); assert.equal(marker.hasToolCall, hasToolCall);
|
|
}
|
|
});
|