import test from 'node:test';
import assert from 'node:assert/strict';
import { mkdtempSync, rmSync, writeFileSync } from 'node:fs';
import { join } from 'node:path';
import { tmpdir } from 'node:os';
import { spawn } from 'node:child_process';
import { once } from 'node:events';
import { startBroker } from '../src/runtime.mjs';
import { Client } from '../src/client.mjs';
import { registerRuntime, updateRuntime, processIdentity } from '../../processing-control/src/runtimes.mjs';
import { readControl } from '../../processing-control/src/state.mjs';

const businesses = { demo: { id: 'demo', human: 'jason', arbiters: { technical: 'coder', delivery: 'coder' },
  roles: { coder: { authority: { withinRole: ['message.send'], crossRole: [] } } } } };
async function broker(root) {
  const r = await startBroker({ dataRoot: root, businesses, readers: ['demo'] });
  const cap = r.broker.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
  return { r, client: new Client({ path: r.path, cap, timeout: 15000 }) };
}

test('real runtime injects the coordinator, stops its workload and mirrors stable stop events across restart', async t => {
  const root = mkdtempSync(join(tmpdir(), 'processing-runtime-bus-'));
  let live;
  const p = spawn(process.execPath, ['-e', "console.log('ready');setInterval(()=>{},1000)"], { detached: true, stdio: ['ignore', 'pipe', 'ignore'] });
  const exit = once(p, 'exit');
  t.after(async () => {
    await live?.close();
    if (p.exitCode === null && p.signalCode === null) { try { process.kill(-p.pid, 'SIGKILL'); } catch {} await exit; }
    rmSync(root, { recursive: true, force: true });
  });
  await once(p.stdout, 'data');
  let current = await broker(root); live = current.r;
  const receipt = registerRuntime(root, { kind: 'session', business: 'demo', run: 'runtime-fixture' });
  updateRuntime(root, receipt.id, { ...processIdentity(p.pid), phase: 'running', group: true });
  const halted = await current.client.call('processing.halt');
  assert.equal(halted.halted, true);
  assert.equal(halted.quiescent, true);
  await exit;
  assert.equal(processIdentity(p.pid), null);
  await assert.rejects(live.mayLaunch({ business: 'demo', instance: 'coder', provider: 'fixture', model: 'fixture' }), { code: 'processing-halted' });
  await live.close(); live = null;
  current = await broker(root); live = current.r;
  assert.equal((await current.client.call('processing.status')).halted, true);
  const events = [halted.generation, receipt.id].flatMap(subject =>
    live.broker.request(live.readers[0].cap, { verb: 'trail', args: { subject } }));
  assert.equal(events.filter(e => e.body?.action === 'processing.halt').length, 1);
  assert.equal(events.filter(e => e.body?.action === 'processing.stop').length, 1);
  assert.equal((await current.client.call('processing.resume', { generation: halted.generation })).halted, false);
});

test('invalid state still starts the control bus and human halt then resume repairs it', async t => {
  const root = mkdtempSync(join(tmpdir(), 'processing-repair-bus-'));
  let live;
  t.after(async () => { await live?.close(); rmSync(root, { recursive: true, force: true }); });
  let current = await broker(root); live = current.r;
  await live.close(); live = null;
  writeFileSync(join(root, 'control', 'processing.json'), '{');
  current = await broker(root); live = current.r;
  assert.equal((await current.client.call('processing.status')).code, 'processing-state-invalid');
  await assert.rejects(current.client.call('processing.resume', { generation: 'bad' }), { code: 'processing-state-invalid' });
  const halted = await current.client.call('processing.halt');
  assert.equal(halted.recovered, true);
  assert.equal((await current.client.call('processing.resume', { generation: halted.generation })).halted, false);
  assert.equal(readControl(root).halted, false);
});