// 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);
  }
});