import test from 'node:test';
import assert from 'node:assert/strict';
import { spawn } from 'node:child_process';
import { once } from 'node:events';
import { mkdtempSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { readControl, assertProcessing } from '../src/state.mjs';
import { registerRuntime, updateRuntime, processIdentity, unregisterRuntime } from '../src/runtimes.mjs';
const { createProcessing } = await import('../src/coordinator.mjs').catch(e => { if (e.code === 'ERR_MODULE_NOT_FOUND') return {}; throw e; });
const actor = { business: 'demo', human: 'jason' };
function fixture(t) {
  assert.equal(typeof createProcessing, 'function');
  const root = mkdtempSync(join(tmpdir(), 'processing-coordinator-'));
  const c = createProcessing({ dataRoot: root, graceMs: 80, killWaitMs: 1000 });
  t.after(async () => { await c.close(); rmSync(root, { recursive: true, force: true }); });
  return { root, c };
}

test('halt publishes before awaiting stops, resume refuses pending, and only clear permits new work', async t => {
  const { root, c } = fixture(t);
  const p = spawn(process.execPath, ['-e', "process.on('SIGTERM',()=>{});console.log('ready');setInterval(()=>{},1000)"], { detached: true, stdio: ['ignore', 'pipe', 'ignore'] });
  const exit = once(p, 'exit');
  t.after(async () => { if (p.signalCode === null && p.exitCode === null) { try { process.kill(-p.pid, 'SIGKILL'); } catch {} await exit; } });
  await once(p.stdout, 'data');
  const r = registerRuntime(root, { business: 'demo', run: 'standin', kind: 'session' });
  updateRuntime(root, r.id, { ...processIdentity(p.pid), group: true, phase: 'running' });
  const halted = c.halt(actor);
  assert.equal(readControl(root).halted, true, 'durable gate is set before any stop await');
  await assert.rejects(c.resume({ ...actor, generation: readControl(root).generation }), { code: 'processing-stop-pending' });
  const result = await halted;
  assert.equal(result.quiescent, true);
  assert.deepEqual(await exit, [null, 'SIGKILL']);
  assert.throws(() => assertProcessing(root), { code: 'processing-halted' });
  const resumed = await c.resume({ ...actor, generation: result.generation });
  assert.equal(resumed.halted, false);
  assert.doesNotThrow(() => assertProcessing(root));
});

test('a new coordinator replays a persisted halt and an unresolved start keeps clear refused', async t => {
  const { root, c } = fixture(t);
  const pending = registerRuntime(root, { business: 'demo', run: 'starting', kind: 'session' });
  const result = await c.halt(actor);
  assert.equal(result.quiescent, false);
  await c.close();
  const next = createProcessing({ dataRoot: root, graceMs: 10, killWaitMs: 10 });
  try {
    assert.equal(next.read().halted, true);
    await assert.rejects(next.resume({ ...actor, generation: result.generation }), { code: 'processing-stop-pending' });
    unregisterRuntime(root, pending.id);
    const resumed = await next.resume({ ...actor, generation: result.generation });
    assert.equal(resumed.halted, false);
  } finally { await next.close(); }
});

test('a new halt generation never inherits a previous quiescent status before its sweep', async t => {
  const { mkdtempSync, rmSync } = await import('node:fs');
  const { tmpdir } = await import('node:os');
  const { join } = await import('node:path');
  const { transition } = await import('../src/state.mjs');
  const root = mkdtempSync(join(tmpdir(), 'processing-counts-'));
  const processing = createProcessing({ dataRoot: root });
  t.after(async () => { await processing.close(); rmSync(root, { recursive: true, force: true }); });
  transition(root, { business: 'demo', human: 'jason', halted: true });
  assert.equal(processing.read().quiescent, false);
  await processing.reconcile(); assert.equal(processing.read().quiescent, true);
});