import test from 'node:test';
import assert from 'node:assert/strict';
import { mkdtempSync, rmSync, readFileSync } from 'node:fs';
import { join } from 'node:path';
import { tmpdir } from 'node:os';
import { initializeControl, readControl, transition } from '../src/state.mjs';
import { registerRuntime, updateRuntime, runtime, stopRun, sweepRuntimes, readStops } from '../src/runtimes.mjs';

const boot = '12345678-1234-4234-8234-123456789abc';
const decision = 'fire-decision';
function setup(t, run = 'unbound-run') {
  const root = mkdtempSync(join(tmpdir(), 'stop-run-')); initializeControl(root);
  t.after(() => rmSync(root, { recursive: true, force: true }));
  const r = registerRuntime(root, { kind: 'session', business: 'demo', run });
  updateRuntime(root, r.id, { pid: 2000000001, startTime: '42', boot, phase: 'running', group: false });
  let time = 0, live = true, changed = false;
  const signals = [], sleeps = [];
  const options = {
    now: () => time,
    sleep: async ms => { sleeps.push(ms); time += ms; },
    identity: pid => live ? { pid, startTime: changed ? '43' : '42', boot } : null,
    state: () => live,
    signal: (_r, sig) => { signals.push(sig); if (sig === 'SIGKILL') live = false; return true; },
  };
  return { root, r, signals, sleeps, options,
    setLive: value => { live = value; }, change: () => { changed = true; },
    args: { root, business: 'demo', run, reason: 'fire', owner: { decision } } };
}
test('unbound run stops by verified registry identity, fixed 5s grace and 2s observe policy', async t => {
  const f = setup(t), before = readControl(f.root);
  const result = await stopRun(f.args, f.options);
  assert.equal(result.outcome, 'killed'); assert.equal(result.pid, 2000000001);
  assert.deepEqual(result.signals, ['SIGTERM', 'SIGKILL']);
  assert.ok(f.sleeps.reduce((a,b)=>a+b,0) >= 5000);
  assert.equal(result.receipt.reason, 'fire'); assert.deepEqual(result.receipt.owner, { decision });
  assert.deepEqual(readControl(f.root), before, 'fire never halts the instance');
});
test('same run joins the identical promise and completed repeats preserve owner and send no signals', async t => {
  const f = setup(t);
  const first = stopRun(f.args, f.options), second = stopRun(f.args, f.options);
  assert.equal(second, first);
  const result = await first;
  const later = await stopRun({ ...f.args, owner: { decision: 'other' } }, f.options);
  assert.deepEqual(later, result); assert.deepEqual(f.signals, ['SIGTERM', 'SIGKILL']);
});
test('identity changed before TERM or between TERM and KILL refuses the changed signal', async t => {
  for (const phase of ['before', 'after-term']) {
    const f = setup(t, phase);
    if (phase === 'before') f.change();
    f.options.signal = (_r, sig) => { f.signals.push(sig); f.change(); return true; };
    const result = await stopRun(f.args, f.options);
    assert.equal(result.outcome, 'identity-changed');
    assert.deepEqual(f.signals, phase === 'before' ? [] : ['SIGTERM']);
    assert.deepEqual(result.signals, f.signals);
  }
});
test('failed signal delivery observes alive, records no invented signals and never restarts escalation', async t => {
  const f = setup(t); f.options.signal = () => false;
  const result = await stopRun(f.args, f.options);
  assert.equal(result.outcome, 'alive'); assert.deepEqual(result.signals, []);
  assert.equal(f.sleeps.reduce((a,b)=>a+b,0), 7000);
  assert.deepEqual(await stopRun(f.args, f.options), result);
});
test('a sleep past the stop deadline still observes an alive run after both signals', async t => {
  const f = setup(t); let time = 0;
  f.options.now = () => time;
  f.options.sleep = async ms => { time += ms + 1; };
  f.options.signal = (_r, sig) => { f.signals.push(sig); return true; };
  const result = await stopRun(f.args, f.options);
  assert.ok(time > 7000, 'the last sleep wakes past the complete stop deadline');
  assert.deepEqual(f.signals, ['SIGTERM', 'SIGKILL']);
  assert.equal(result.outcome, 'alive');
  assert.deepEqual(result.signals, ['SIGTERM', 'SIGKILL']);
});
test('missing run and another business never borrow a pid or signal', async t => {
  const f = setup(t);
  for (const args of [{ ...f.args, run: 'missing' }, { ...f.args, business: 'other' }]) {
    const result = await stopRun(args, f.options);
    assert.equal(result.outcome, 'unknown'); assert.equal(result.pid, null); assert.deepEqual(result.signals, []);
  }
  assert.deepEqual(f.signals, []);
});
test('halt joins an in-progress fire without another TERM or deadline and links its first receipt', async t => {
  const f = setup(t); let entered, release;
  const ready = new Promise(resolve => { entered = resolve; });
  const gate = new Promise(resolve => { release = resolve; });
  let blocked = false;
  const sleep = f.options.sleep;
  f.options.sleep = async ms => { if (!blocked) { blocked = true; entered(); await gate; } await sleep(ms); };
  const first = stopRun(f.args, f.options); await ready;
  const halted = transition(f.root, { business: 'demo', human: 'jason', halted: true });
  const sweep = sweepRuntimes(f.root, halted.generation, { stopOptions: f.options });
  release(); const result = await first, summary = await sweep;
  assert.equal(summary.quiescent, true); assert.deepEqual(f.signals, ['SIGTERM', 'SIGKILL']);
  const [link] = readStops(f.root, halted.generation);
  assert.equal(link.stop, result.receipt.id); assert.equal(link.reason, 'fire');
  assert.equal(runtime(f.root, f.r.id).haltGeneration, undefined, 'halt does not relabel a fire-owned stop');
  assert.deepEqual(link.owner, { decision });
});

test('a separate host joins durable first ownership without duplicate signals', async t => {
  const { spawn } = await import('node:child_process');
  const { once } = await import('node:events');
  const f = setup(t), source = new URL('../src/runtimes.mjs', import.meta.url).href;
  const script = `import {stopRun} from ${JSON.stringify(source)};
    let time=0,live=true,waiting=true;
    const args=JSON.parse(process.argv[1]);
    const result=await stopRun(args,{
      now:()=>time, identity:pid=>live?{pid,startTime:'42',boot:${JSON.stringify(boot)}}:null,
      state:()=>live,signal:(_r,sig)=>{process.send({signal:sig});if(sig==='SIGKILL')live=false;return true},
      sleep:async ms=>{if(waiting){waiting=false;await new Promise(r=>process.once('message',r))}time+=ms}
    });process.send({result});process.disconnect();`;
  const child = spawn(process.execPath, ['--input-type=module', '-e', script, JSON.stringify(f.args)],
    { detached: true, stdio: ['ignore', 'pipe', 'pipe', 'ipc'] });
  t.after(() => { if (child.exitCode === null) { try { process.kill(-child.pid, 'SIGKILL'); } catch {} } });
  const messages = []; child.on('message', value => messages.push(value));
  await once(child, 'message');
  const joined = stopRun({ ...f.args, owner: { decision: 'second-host' } },
    { ...f.options, now: Date.now, sleep: ms => new Promise(resolve => setTimeout(resolve, ms)) });
  setTimeout(() => child.send('continue'), 30);
  const result = await joined;
  const exit = child.exitCode === null ? await once(child, 'exit') : [child.exitCode];
  assert.equal(exit[0], 0);
  assert.deepEqual(f.signals, [], 'the second host sends no signal');
  assert.deepEqual(messages.filter(m => m.signal).map(m => m.signal), ['SIGTERM', 'SIGKILL']);
  assert.equal(result.receipt.id, messages.find(m => m.result).result.receipt.id);
  assert.deepEqual(result.receipt.owner, { decision });
});

test('corrupt and symlinked final receipts refuse as unknown rather than signalling again', async t => {
  const { readdirSync, writeFileSync, symlinkSync, unlinkSync } = await import('node:fs');
  const f = setup(t); const complete = await stopRun(f.args, f.options);
  const dir = join(f.root, 'control', 'run-stops');
  const name = readdirSync(dir).find(n => n.endsWith('.result.json'));
  writeFileSync(join(dir, name), '{broken', { mode: 0o600 });
  assert.equal((await stopRun(f.args, f.options)).outcome, 'unknown');
  writeFileSync(join(dir, name), JSON.stringify({ ...complete.receipt, outcome: 'invented' }), { mode: 0o600 });
  assert.equal((await stopRun(f.args, f.options)).outcome, 'unknown');
  const target = join(f.root, 'outside'); writeFileSync(target, JSON.stringify(complete.receipt), { mode: 0o600 });
  unlinkSync(join(dir, name)); symlinkSync(target, join(dir, name));
  assert.equal((await stopRun(f.args, f.options)).outcome, 'unknown');
  assert.deepEqual(f.signals, ['SIGTERM', 'SIGKILL']);
});

test('a later exit gets separate halt evidence while the original alive result stays immutable', async t => {
  const { spawn } = await import('node:child_process');
  const { once } = await import('node:events');
  const { processIdentity } = await import('../src/runtimes.mjs');
  const f = setup(t);
  const child = spawn(process.execPath, ['-e', "console.log('ready');setInterval(()=>{},1000)"],
    { detached: true, stdio: ['ignore', 'pipe', 'pipe'] });
  t.after(() => { try { process.kill(-child.pid, 'SIGKILL'); } catch {} });
  await once(child.stdout, 'data');
  const identity = processIdentity(child.pid);
  updateRuntime(f.root, f.r.id, { ...identity, group: false });
  f.options.identity = () => identity; f.options.signal = () => false;
  const first = await stopRun(f.args, f.options);
  assert.equal(first.outcome, 'alive');
  const bytes = JSON.stringify(first);
  const ended = once(child, 'exit'); child.kill('SIGKILL'); await ended;
  const state = transition(f.root, { business: 'demo', human: 'jason', halted: true });
  assert.equal((await sweepRuntimes(f.root, state.generation)).quiescent, true);
  assert.equal(JSON.stringify(await stopRun(f.args, f.options)), bytes);
  const { readdirSync } = await import('node:fs');
  assert.equal(readdirSync(join(f.root, 'control', 'run-stops')).filter(n => n.endsWith('.observation.json')).length, 1);
});

test('root aliases join but a different root never shares the run stop', async t => {
  const { symlinkSync } = await import('node:fs');
  const a = setup(t), b = setup(t);
  const alias = a.root + '-alias'; symlinkSync(a.root, alias); t.after(() => rmSync(alias));
  const first = stopRun(a.args, a.options);
  assert.equal(stopRun({ ...a.args, root: alias }, a.options), first);
  const other = stopRun(b.args, b.options);
  assert.notEqual(other, first);
  const [one, two] = await Promise.all([first, other]);
  assert.notEqual(one.receipt.id, two.receipt.id);
  assert.deepEqual(a.signals, ['SIGTERM', 'SIGKILL']); assert.deepEqual(b.signals, ['SIGTERM', 'SIGKILL']);
});

test('a changed runtime record cannot replace the original stop identity after TERM', async t => {
  const f = setup(t);
  f.options.signal = (_r, sig) => {
    f.signals.push(sig); f.change(); updateRuntime(f.root, f.r.id, { startTime: '43' }); return true;
  };
  const result = await stopRun(f.args, f.options);
  assert.equal(result.outcome, 'identity-changed'); assert.deepEqual(f.signals, ['SIGTERM']);
});
test('identity is rechecked at actual signal delivery, not just at the loop observation', async t => {
  const f = setup(t); let reads = 0;
  f.options.identity = pid => ({ pid, boot, startTime: ++reads === 1 ? '42' : '43' });
  const result = await stopRun(f.args, f.options);
  assert.equal(result.outcome, 'identity-changed'); assert.deepEqual(f.signals, []);
});