import test from 'node:test';
import assert from 'node:assert/strict';
import { spawn, spawnSync } from 'node:child_process';
import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { once } from 'node:events';
import { initializeControl, transition } from '../src/state.mjs';
const api = await import('../src/runtimes.mjs').catch(e => { if (e.code === 'ERR_MODULE_NOT_FOUND') return {}; throw e; });
const actor = { business: 'demo', human: 'jason', halted: true };
const delay = ms => new Promise(resolve => setTimeout(resolve, ms));
function fixture(t) {
  const root = mkdtempSync(join(tmpdir(), 'processing-runtime-'));
  initializeControl(root); t.after(() => rmSync(root, { recursive: true, force: true }));
  return root;
}
async function child(t, stubborn = false) {
  const p = spawn(process.execPath, ['-e', `process.on('SIGTERM',()=>{${stubborn ? '' : 'process.exit(0)'}});console.log('ready');setInterval(()=>{},1000)`], { detached: true, stdio: ['ignore', 'pipe', 'pipe'] });
  const ended = once(p, 'exit');
  t.after(async () => { if (p.exitCode === null && p.signalCode === null) { try { process.kill(-p.pid, 'SIGKILL'); } catch {} await ended; } });
  await once(p.stdout, 'data');
  return { p, ended };
}
function receipt(root, p) {
  const r = api.registerRuntime(root, { kind: 'session', business: 'demo', run: 'run' });
  return api.updateRuntime(root, r.id, { ...api.processIdentity(p.pid), phase: 'running', group: true });
}

// Without TERM, a cooperative child would survive until unnecessary escalation.
test('halt terminates a tracked stand-in and leaves honest write-once stop evidence', async t => {
  assert.equal(typeof api.registerRuntime, 'function');
  const root = fixture(t), { p, ended } = await child(t);
  const r = receipt(root, p);
  const evidence = join(root, 'evidence.json'); writeFileSync(evidence, 'immutable existing result');
  const state = transition(root, actor);
  const result = await api.sweepRuntimes(root, state.generation, { graceMs: 150, killWaitMs: 1000 });
  assert.equal(result.quiescent, true);
  assert.equal(result.remaining, 0);
  assert.deepEqual(await ended, [0, null]);
  const stops = api.readStops(root, state.generation);
  assert.equal(stops.length, 1);
  assert.equal(stops[0].runtime, r.id);
  assert.equal(stops[0].outcome, 'terminated');
  assert.deepEqual(stops[0].signals, ['SIGTERM']);
  assert.equal(readFileSync(evidence, 'utf8'), 'immutable existing result');
  const bytes = JSON.stringify(stops);
  await api.sweepRuntimes(root, state.generation, { graceMs: 10, killWaitMs: 1000 });
  assert.equal(JSON.stringify(api.readStops(root, state.generation)), bytes);
});

// Dropping the KILL escalation leaves this actual child alive and fails quiescence.
test('a TERM-resistant workload is killed after the bounded wait', async t => {
  assert.equal(typeof api.registerRuntime, 'function');
  const root = fixture(t), { p, ended } = await child(t, true);
  receipt(root, p);
  const state = transition(root, actor), begin = Date.now();
  const result = await api.sweepRuntimes(root, state.generation, { graceMs: 80, killWaitMs: 1000 });
  assert.equal(result.quiescent, true);
  assert.deepEqual(await ended, [null, 'SIGKILL']);
  assert.ok(Date.now() - begin >= 80);
  assert.deepEqual(api.readStops(root, state.generation)[0].signals, ['SIGTERM', 'SIGKILL']);
});

// An unchecked stale start identity would terminate a different process.
test('wrong start ticks or boot identity cannot signal a live replacement', async t => {
  assert.equal(typeof api.registerRuntime, 'function');
  const root = fixture(t), { p } = await child(t);
  const r = receipt(root, p);
  api.updateRuntime(root, r.id, { startTime: '0' });
  const state = transition(root, actor);
  await api.sweepRuntimes(root, state.generation, { graceMs: 30, killWaitMs: 30 });
  await delay(30);
  assert.equal(p.exitCode, null); assert.equal(p.signalCode, null);
  assert.equal(api.readStops(root, state.generation)[0].outcome, 'unknown');
});

test('a pending start cannot be reported quiescent and a second root is never signalled', async t => {
  assert.equal(typeof api.registerRuntime, 'function');
  const a = fixture(t), b = fixture(t), { p } = await child(t);
  api.registerRuntime(a, { kind: 'session', business: 'demo', run: 'pending' });
  receipt(b, p);
  const state = transition(a, actor);
  const result = await api.sweepRuntimes(a, state.generation, { graceMs: 30, killWaitMs: 30 });
  assert.equal(result.quiescent, false);
  assert.equal(result.remaining, 1);
  assert.equal(p.exitCode, null); assert.equal(p.signalCode, null);
});

// A group leader can exit on TERM while its child ignores it. Quiescence is the whole group.
test('escalation still kills the tracked group after its leader exits', async t => {
  const root = fixture(t);
  const script = `const {spawn}=require('node:child_process');
    const c=spawn(process.execPath,['-e',"process.on('SIGTERM',()=>{});console.log('ready');setInterval(()=>{},1000)"],{stdio:['ignore','pipe','inherit']});
    c.stdout.once('data',()=>console.log(c.pid));process.on('SIGTERM',()=>process.exit(0));setInterval(()=>{},1000);`;
  const p = spawn(process.execPath, ['-e', script], { detached: true, stdio: ['ignore', 'pipe', 'pipe'] });
  t.after(() => { try { process.kill(-p.pid, 'SIGKILL'); } catch {} });
  const [line] = await once(p.stdout, 'data');
  const descendant = Number(String(line).trim());
  const r = receipt(root, p), state = transition(root, actor);
  const outcome = await api.sweepRuntimes(root, state.generation, { graceMs: 80, killWaitMs: 1000 });
  assert.equal(outcome.quiescent, true);
  assert.equal(api.processIdentity(descendant), null, 'stubborn descendant must not survive the group leader');
  assert.equal(api.readStops(root, state.generation).find(s => s.runtime === r.id).outcome, 'killed');
});

// Session TERM goes to the runner, leaving the outer namespace for forced escalation.
test('session TERM targets its verified runner before forcing the outer namespace', async t => {
  const root = fixture(t);
  const script = `const {spawn}=require('node:child_process');
    process.on('SIGTERM',()=>require('node:fs').writeFileSync(${JSON.stringify(join(root, 'outer-term'))},'unexpected'));
    const c=spawn(process.execPath,['-e',"process.on('SIGTERM',()=>process.exit(0));console.log('ready');setInterval(()=>{},1000)"],{stdio:['ignore','pipe','inherit']});
    c.stdout.once('data',()=>console.log(c.pid));c.once('exit',()=>process.exit(0));setInterval(()=>{},1000);`;
  const p = spawn(process.execPath, ['-e', script], { detached: true, stdio: ['ignore', 'pipe', 'pipe'] });
  t.after(() => { try { process.kill(-p.pid, 'SIGKILL'); } catch {} });
  const [line] = await once(p.stdout, 'data'), runner = api.processIdentity(Number(String(line).trim()));
  const r = receipt(root, p);
  api.updateRuntime(root, r.id, { runnerPid: runner.pid, runnerStartTime: runner.startTime });
  const state = transition(root, actor);
  const result = await api.sweepRuntimes(root, state.generation, { graceMs: 300, killWaitMs: 1000 });
  assert.equal(result.quiescent, true);
  assert.deepEqual(api.readStops(root, state.generation)[0].signals, ['SIGTERM']);
  assert.equal(api.runtime(root, r.id).haltGeneration, state.generation);
  assert.equal(existsSync(join(root, 'outer-term')), false);
});

// Publishing a receipt is not durable if its newly-created containing directory can vanish.
test('new runtime and stop directories sync their containing directories', t => {
  const root = fixture(t);
  const script = `import fs from 'node:fs';import {syncBuiltinESMExports} from 'node:module';
    const real=fs.fsyncSync,seen=[];fs.fsyncSync=fd=>{seen.push(fs.readlinkSync('/proc/self/fd/'+fd));return real(fd)};
    syncBuiltinESMExports();const api=await import(process.argv[1]);
    const {transition}=await import(process.argv[2]);
    const r=api.registerRuntime(process.argv[3],{kind:'task',business:'demo',run:'gone'});
    api.updateRuntime(process.argv[3],r.id,{phase:'ended'});
    const s=transition(process.argv[3],{business:'demo',human:'jason',halted:true});
    await api.sweepRuntimes(process.argv[3],s.generation);
    console.log(JSON.stringify({seen,generation:s.generation}));`;
  const child = spawnSync(process.execPath, ['--input-type=module', '-e', script,
    new URL('../src/runtimes.mjs', import.meta.url).href, new URL('../src/state.mjs', import.meta.url).href, root], { encoding: 'utf8' });
  assert.equal(child.status, 0, child.stderr);
  const { seen, generation } = JSON.parse(child.stdout);
  assert.ok(seen.filter(p => p === join(root, 'control')).length >= 3, 'runtime mkdir, halt rename and stops mkdir sync control');
  assert.ok(seen.includes(join(root, 'control', 'stops')), 'generation directory creation syncs stops');
  assert.ok(seen.includes(join(root, 'control', 'stops', generation)), 'stop receipt syncs generation directory');
});

test('a vanished group leader with unverified surviving members stays unknown, never quiescent', async t => {
  const root = fixture(t);
  const script = `const {spawn}=require('node:child_process');const c=spawn(process.execPath,['-e',"console.log('ready');setInterval(()=>{},1000)"],{stdio:['ignore','pipe','inherit']});c.stdout.once('data',()=>console.log(c.pid));process.on('SIGTERM',()=>process.exit(0));setInterval(()=>{},1000)`;
  const p = spawn(process.execPath, ['-e', script], { detached: true, stdio: ['ignore', 'pipe', 'pipe'] });
  t.after(() => { try { process.kill(-p.pid, 'SIGKILL'); } catch {} });
  const [line] = await once(p.stdout, 'data'); const descendant = Number(String(line).trim());
  receipt(root, p); const exited = once(p, 'exit'); p.kill('SIGTERM'); await exited;
  const state = transition(root, actor);
  const result = await api.sweepRuntimes(root, state.generation, { graceMs: 20, killWaitMs: 30 });
  assert.equal(result.quiescent, false); assert.equal(result.remaining, 1);
  assert.notEqual(api.processIdentity(descendant), null, 'never signal a group with no verified surviving anchor');
});
test('persisted member identity anchors a group after the leader exited before the sweep', async t => {
  const root = fixture(t);
  const script = `const {spawn}=require('node:child_process');const c=spawn(process.execPath,['-e',"process.on('SIGTERM',()=>{});console.log('ready');setInterval(()=>{},1000)"],{stdio:['ignore','pipe','inherit']});c.stdout.once('data',()=>console.log(c.pid));process.on('SIGTERM',()=>process.exit(0));setInterval(()=>{},1000)`;
  const p = spawn(process.execPath, ['-e', script], { detached: true, stdio: ['ignore', 'pipe', 'pipe'] });
  t.after(() => { try { process.kill(-p.pid, 'SIGKILL'); } catch {} });
  const [line] = await once(p.stdout, 'data'), descendant = Number(String(line).trim()), r = receipt(root, p);
  api.updateRuntime(root, r.id, { members: [api.processIdentity(descendant)] });
  const exited = once(p, 'exit'); p.kill('SIGTERM'); await exited;
  const state = transition(root, actor);
  const result = await api.sweepRuntimes(root, state.generation, { graceMs: 30, killWaitMs: 1000 });
  assert.equal(result.quiescent, true); assert.equal(api.processIdentity(descendant), null);
  assert.deepEqual(api.readStops(root, state.generation)[0].signals, ['SIGTERM', 'SIGKILL']);
});

test('an unconfirmed secondary brake does not leave a verified workload running or claim success', async t => {
  const root = fixture(t), { p, ended } = await child(t);
  receipt(root, p); const state = transition(root, actor);
  const result = await api.sweepRuntimes(root, state.generation, { graceMs: 60, killWaitMs: 1000,
    beforeStop: () => { throw Error('fixture journal unavailable'); } });
  await ended;
  assert.equal(api.processIdentity(p.pid), null);
  assert.equal(result.quiescent, false); assert.equal(result.remaining, 1);
  const stops = api.readStops(root, state.generation); assert.equal(stops[0].outcome, 'unknown');
  assert.equal(stops[0].problem, 'brake-unconfirmed');
});