import test from 'node:test';
import assert from 'node:assert/strict';
import { spawnSync } from 'node:child_process';
import { mkdirSync, readFileSync, writeFileSync } from 'node:fs';
import { join } from 'node:path';
import { fileURLToPath } from 'node:url';
import { fixture, tmp, withoutHostCredentials, fakeHarnesses } from './helpers.mjs';
import { bootConfig, loadSystem } from '../src/config.mjs';
import { startHost, startTimeOf } from '../src/host.mjs';
import { createLauncher } from '../src/launcher.mjs';
import { Client } from '../../bus/src/client.mjs';
import { gitEnv } from '../../seat/src/checkout.mjs';

async function setup(t, edit = d => d, { revoked = false } = {}) {
  const root = tmp(t, 'request-launch-'), f = fixture(root, 'demo', edit);
  withoutHostCredentials(f.env, { piKey: true });
  const repo = f.doc.projects.stack.root;
  for (const args of [['init','-qb','main'],['-c','user.name=test','-c','user.email=test@example.invalid','commit','--allow-empty','-qm','seed']]) {
    const r = spawnSync('git', ['-C', repo, ...args], { env: gitEnv(), encoding: 'utf8' }); assert.equal(r.status, 0, r.stderr);
  }
  const fake = fileURLToPath(new URL('../../harness/tests/fixtures/fake-adapter.mjs', import.meta.url));
  const adapter = join(root, 'adapter.sh'); writeFileSync(adapter, `exec '${process.execPath}' '${fake}'\n`);
  const makeLoader = await fakeHarnesses(t, { pi: adapter, 'claude-code': adapter });
  const system = loadSystem({ env: f.env }), host = await startHost({ boot: bootConfig({ system, businessId: 'demo', env: f.env }), business: 'demo' });
  const launcher = createLauncher({ host, system, env: f.env, makeLoader, harnessVersion: 'fixture', readOnly: [fake], sessionDefaults: { pollInterval: 30 } });
  await launcher.listen();
  t.after(() => host.close(0));
  if (revoked) {
    const { DatabaseSync } = await import('node:sqlite');
    const db = new DatabaseSync(join(f.dataRoot, 'bus/bus.sqlite'));
    db.prepare("INSERT INTO events(id,at,business,kind,subject,body) VALUES('fixture-revoked',?,'demo','launch.revoked',NULL,'{}')").run(new Date().toISOString());
    db.close();
  }
  const bound = {};
  for (const role of ['chief','pm']) {
    bound[role] = await host.bindLaunch({ business: 'demo', role, run: role+'-fixture', harness: 'pi', pid: process.pid, startTime: startTimeOf(process.pid) });
    const c = new Client({ path: host.path, cap: bound[role].cap }); await c.call('role.claim');
  }
  return { root, f, host, launcher, bound };
}

test('PM cannot use continuation launch recovery to gain role.launch', async t => {
  const x = await setup(t);
  const out = await x.launcher.fromSession(x.bound.pm.cap, 'coder');
  assert.equal(out.ok, false);
  assert.ok(['decision-required','not-authorized'].includes(out.error), out.error);
});

test('chief launch refusals are returned unchanged and stored through the host request path', async t => {
  const x = await setup(t);
  const c = new Client({ path: x.host.path, cap: x.bound.pm.cap });
  // Host operations are not broker verbs and cannot be invoked by a managed session.
  await assert.rejects(c.call('request.launchResult', { instance: 'coder', code: 'capacity-full' }), /unknown-verb/);
  const out = await x.launcher.fromSession(x.bound.chief.cap, 'absent');
  assert.equal(out.error, 'instance-not-listed');
  const pending = await x.host.op({ op: 'request.launches', business: 'demo' });
  assert.deepEqual(pending, []);
});

async function until(fn) {
  const end = Date.now() + 10000;
  while (Date.now() < end) { const value = await fn(); if (value) return value; await new Promise(r => setTimeout(r, 25)); }
  throw Error('request launch did not settle');
}

test('the real chief launches an existing worker with the admitted project in its manifest', async t => {
  const x = await setup(t);
  const out = await x.launcher.fromSession(x.bound.chief.cap, 'coder');
  assert.equal(out.ok, true, out.error);
  const session = x.launcher.registry.sessions.find(s => s.run === out.result.run);
  const manifest = JSON.parse(readFileSync(join(session.runDir, 'bundle/manifest.json')));
  assert.deepEqual(manifest.projects, ['stack']);
  assert.equal(manifest.definition, 'coder');
  assert.ok(manifest.typed.includes('push_working'));
  await x.host.close(0);
});

test('revoked launch and full model limits preserve their existing refusal codes', async t => {
  for (const code of ['launch-revoked','capacity-full']) await t.test(code, async t => {
    const x = await setup(t, doc => {
      if (code === 'capacity-full') doc.launch.limits.models.zai['glm-5.3-flash'] = 0;
      return doc;
    }, { revoked: code === 'launch-revoked' });
    const out = await x.launcher.fromSession(x.bound.chief.cap, 'coder');
    assert.equal(out.error, code);
    assert.equal(x.launcher.registry.sessions.length, 0);
  });
});

function requestHost(x) {
  const operations = {
      identity: m => x.bus.identity(m.cap), authorizeLaunch: m => x.bus.authorizeLaunch(m.cap, m.instance),
      refuse: m => x.bus.refuse(m.cap, m.code), credentialStatus: m => x.bus.credentialStatus(m.business, m.role),
      modelPreflight: m => x.bus.modelPreflight(m), mayLaunch: m => x.bus.mayLaunch(m),
      landHeld: m => x.bus.landHeld(m.record), landHolders: m => x.bus.landHolders(m.business),
      sessionClears: m => x.bus.sessionClears(m.business),
      threadsState: m => x.bus.threadsState(m.business), lifecycleRequests: m => x.bus.lifecycleRequests(m.business),
      'request.admission': m => x.bus.admission(m), 'request.launches': m => x.bus.requestLaunches(m.business),
      'request.launchResult': m => x.bus.requestLaunchResult(m.record), 'request.manifest': m => x.bus.requestManifest(m.record),
      'request.release': m => x.bus.requestRelease(m.cap, m.record), 'request.chief': m => x.bus.requestChief(m.business),
      endLaunch: m => x.bus.endLaunch(m.record),
    };
    return { business: 'demo', path: x.bus.path, dataRoot: x.f.dataRoot, beforeClose: () => {},
      op: m => Promise.resolve().then(() => {
        assert.ok(operations[m.op], `unexpected host operation ${m.op}`); return operations[m.op](m);
      }),
      bindLaunch: record => { const out = x.bus.bindLaunch(record); x.caps[record.role] = out.cap; return out; } };
}

test('an old product manifest drains its active turn before the chief relaunches through normal admission', { timeout: 30000 }, async t => {
  const { requestFixture } = await import('../../bus/tests/request-fixture.mjs');
  let launcher, worker, host, entered, released;
  const f = await requestFixture(t, 'autonomous', async x => {
    withoutHostCredentials(x.f.env, { piKey: true });
    const adapter = join(x.f.dataRoot, 'hold.sh'), source = join(x.f.dataRoot, 'hold.mjs');
    writeFileSync(source, `import {writeFileSync,existsSync} from 'node:fs';
const marker=process.env.MOSAIC_TURN_MARKER;
if ((process.env.MOSAIC_REQUEST??'').includes('HOLD-REQUEST-TURN')) {
 writeFileSync(marker+'.entered','entered');
 while(!existsSync(marker+'.release')) await new Promise(r=>setTimeout(r,25));
}
writeFileSync(marker,'done'); process.stdout.write('turn completed');`);
    writeFileSync(adapter, `exec '${process.execPath}' '${source}'\n`);
    const makeLoader = await fakeHarnesses(t, { pi: adapter, 'claude-code': adapter });
    host = requestHost(x);
    launcher = createLauncher({ host, system: x.system, env: x.f.env, makeLoader, harnessVersion: 'fixture', readOnly: [source], sessionDefaults: { pollInterval: 25 } });
    t.after(() => launcher.close());
    x.bus.broker.endLaunch({ business: 'demo', run: 'pm-run', reason: 'exit', exitCode: 0 });
    const out = await launcher.fromSession(x.caps.chief, 'pm'); assert.equal(out.ok, true, out.error);
    worker = launcher.registry.sessions.find(s => s.run === out.result.run);
    entered = join(worker.runDir, 'turn.done.entered'); released = join(worker.runDir, 'turn.done.release');
    await until(async () => { try { x.call('pm', 'state'); return true; } catch { return false; } });
    x.call('human', 'message.send', { to: 'pm', body: 'HOLD-REQUEST-TURN' });
    const { existsSync } = await import('node:fs'); await until(() => existsSync(entered));
  });
  const original = JSON.parse(readFileSync(join(worker.runDir, 'bundle/manifest.json')));
  assert.deepEqual(original.projects, ['stack']);
  await f.bus.requestLifecycle.recover();
  await launcher.tick();
  const chief = f.call('chief', 'message.receive');
  assert.ok(chief.some(m => m.body.includes('lacks admitted project timer')));
  const refused = await launcher.fromSession(f.caps.chief, 'pm');
  assert.equal(refused.error, 'request-manifest-draining');
  assert.ok(launcher.registry.sessions.some(s => s.run === worker.run), 'active turn remains alive');
  assert.equal((await f.call('human', 'request.get', { request: f.request })).launch_refusal.code, 'request-manifest-draining');
  writeFileSync(released, 'release');
  // The runtime keeps the same drain intent while the turn finishes; the chief path owns the release.
  await until(async () => { await launcher.tick(); return !launcher.registry.sessions.some(s => s.run === worker.run); });
  const next = await launcher.fromSession(f.caps.chief, 'pm'); assert.equal(next.ok, true, next.error);
  const successor = launcher.registry.sessions.find(s => s.run === next.result.run);
  assert.deepEqual(JSON.parse(readFileSync(join(successor.runDir, 'bundle/manifest.json'))).projects, ['stack','timer']);
  const { existsSync } = await import('node:fs');
  assert.equal(existsSync(join(f.f.dataRoot, 'workspaces/demo/pm/timer/.git')), true);
  await launcher.close();
});

test('an unavailable chief records its exact refusal, stays quiet, and retries the same path on admission revision change', { timeout: 30000 }, async t => {
  const { requestFixture } = await import('../../bus/tests/request-fixture.mjs');
  const f = await requestFixture(t); withoutHostCredentials(f.f.env, { piKey: true });
  const { dirname } = await import('node:path');
  const config = join(dirname(f.f.config), 'businesses/demo.json');
  const doc = JSON.parse(readFileSync(config));
  doc.launch.limits.models.zai['glm-5.3-flash'] = 0; writeFileSync(config, JSON.stringify(doc));
  f.bus.broker.endLaunch({ business: 'demo', run: 'chief-run', reason: 'exit', exitCode: 0 });
  const fake = fileURLToPath(new URL('../../harness/tests/fixtures/fake-adapter.mjs', import.meta.url));
  const adapter = join(f.f.dataRoot, 'chief.sh'); writeFileSync(adapter, `exec '${process.execPath}' '${fake}'\n`);
  const makeLoader = await fakeHarnesses(t, { pi: adapter, 'claude-code': adapter });
  const launcher = createLauncher({ host: requestHost({ bus: f.bus, f: f.f, caps: f.caps }), system: f.system, env: f.f.env, makeLoader, harnessVersion: 'fixture', readOnly: [fake], sessionDefaults: { pollInterval: 25 } });
  t.after(() => launcher.close());
  await launcher.listen();
  await f.bus.requestLifecycle.recover(); await launcher.tick();
  assert.equal((await f.call('pm', 'request.get', { request: f.request })).launch_refusal.code, 'capacity-full');
  const attempts = f.journal.effects('demo', f.request).filter(e => e.result?.type === 'launch').length;
  for (let n = 0; n < 3; n++) await launcher.tick();
  assert.equal(f.journal.effects('demo', f.request).filter(e => e.result?.type === 'launch').length, attempts);
  doc.launch.limits.models.zai['glm-5.3-flash'] = 1; writeFileSync(config, JSON.stringify(doc));
  await launcher.tick();
  assert.equal(launcher.registry.sessions.length, 1, JSON.stringify(await f.call('pm', 'request.get', { request: f.request })));
  assert.equal(launcher.registry.sessions[0].instance, 'chief');
  assert.equal((await f.call('pm', 'request.get', { request: f.request })).launch_refusal, null);
  await launcher.close();
});

test('archived product mappings stay out of new manifests while the business still serves live projects', async t => {
  const { requestFixture } = await import('../../bus/tests/request-fixture.mjs');
  const x = await requestFixture(t);
  withoutHostCredentials(x.f.env, { piKey: true });
  const project = x.store.get("SELECT id FROM projects WHERE repo='timer'");
  assert.ok(project);
  await x.call('human', 'project.archive', { project: project.id });
  const fake = fileURLToPath(new URL('../../harness/tests/fixtures/fake-adapter.mjs', import.meta.url));
  const adapter = join(x.f.dataRoot, 'archive-launch.sh'); writeFileSync(adapter, `exec '${process.execPath}' '${fake}'\n`);
  const launcher = createLauncher({ host: requestHost({ bus: x.bus, f: x.f, caps: x.caps }), system: x.system, env: x.f.env,
    makeLoader: await fakeHarnesses(t, { pi: adapter, 'claude-code': adapter }), harnessVersion: 'fixture', readOnly: [fake], sessionDefaults: { pollInterval: 25 } });
  t.after(() => launcher.close()); await launcher.listen();
  const out = await launcher.fromSession(x.caps.chief, 'coder');
  assert.equal(out.ok, true, out.error);
  const session = launcher.registry.sessions.find(s => s.run === out.result.run);
  assert.deepEqual(JSON.parse(readFileSync(join(session.runDir, 'bundle/manifest.json'))).projects, ['stack']);
  await launcher.close();
});

// A resting request has an outstanding PM continuation and no running chief. These
// tests use the same broker throughout; only the launcher is replaced for --no-launch.
async function restingRequest(t, { launches = true, brokenHolders = false } = {}) {
  let launcher;
  // Close the launcher before requestFixture closes its bus and removes its root.
  t.after(async () => { if (launcher) await launcher.close(); });
  const { requestFixture } = await import('../../bus/tests/request-fixture.mjs');
  const f = await requestFixture(t);
  withoutHostCredentials(f.f.env, { piKey: true });
  f.bus.broker.endLaunch({ business: 'demo', run: 'chief-run', reason: 'exit', exitCode: 0 });
  await f.bus.requestLifecycle.recover();
  const fake = fileURLToPath(new URL('../../harness/tests/fixtures/fake-adapter.mjs', import.meta.url));
  const adapter = join(f.f.dataRoot, 'resume-chief.sh');
  writeFileSync(adapter, `exec '${process.execPath}' '${fake}'\n`);
  // The real bus host creates this before constructing its read-only launcher.
  mkdirSync(join(f.f.dataRoot, 'bus-host'), { recursive: true, mode: 0o700 });
  const makeLoader = await fakeHarnesses(t, { pi: adapter, 'claude-code': adapter });
  const host = requestHost({ bus: f.bus, f: f.f, caps: f.caps });
  const op = host.op;
  let chiefChecks = 0, inChiefLaunch = false;
  host.op = m => {
    if (m.op === 'request.chief') { chiefChecks++; inChiefLaunch = true; }
    if (brokenHolders && inChiefLaunch && m.op === 'landHolders') return Promise.resolve({ malformed: true });
    return op(m);
  };
  async function start(enabled) {
    launcher = createLauncher({ host, system: f.system, env: f.f.env, launches: enabled, makeLoader,
      harnessVersion: 'fixture', readOnly: [fake], threadTickMs: 3600000, sessionDefaults: { pollInterval: 25 } });
    await launcher.listen();
    return launcher;
  }
  await start(launches);
  return { f, host, start, get launcher() { return launcher; }, chiefChecks: () => chiefChecks,
    async replace(enabled = true) { await launcher.close(); launcher = null; return start(enabled); } };
}

async function expectQuietTicks(x, code) {
  await x.launcher.tick();
  const { f } = x;
  assert.equal((await f.call('human', 'request.get', { request: f.request })).launch_refusal.code, code);
  const effects = f.journal.effects('demo', f.request).length;
  const messages = f.store.get('SELECT count(*) n FROM messages').n;
  for (let i = 0; i < 5; i++) await x.launcher.tick();
  assert.equal(x.launcher.registry.sessions.length, 0, 'ticks do not launch');
  assert.equal(x.chiefChecks(), 0, 'ticks never enter chief launch');
  assert.equal(f.journal.effects('demo', f.request).length, effects, 'identical refusal receipts are deduplicated');
  assert.equal(f.store.get('SELECT count(*) n FROM messages').n, messages, 'ticks do not flood messages');
}

test('a resting chief launches on a later tick after processing resumes', { timeout: 30000 }, async t => {
  const x = await restingRequest(t), { f } = x;
  const halted = await f.call('human', 'processing.halt');
  await expectQuietTicks(x, 'processing-halted');
  await f.call('human', 'processing.resume', { generation: halted.generation });
  await x.launcher.tick();
  assert.equal(x.launcher.registry.sessions.length, 1, 'resume must launch without restarting the bus');
  assert.equal(x.launcher.registry.sessions[0].instance, 'chief');
  assert.equal((await f.call('human', 'request.get', { request: f.request })).launch_refusal, null);
});

test('a no-launch host remains quiet and an enabled launcher can subsequently start the chief', { timeout: 30000 }, async t => {
  const x = await restingRequest(t, { launches: false });
  await expectQuietTicks(x, 'launches-disabled');
  await x.replace();
  await x.launcher.tick();
  assert.equal(x.launcher.registry.sessions.length, 1, 'enabling launches must not require a bus restart');
  assert.equal(x.launcher.registry.sessions[0].instance, 'chief');
  assert.equal((await x.f.call('human', 'request.get', { request: x.f.request })).launch_refusal, null);
});

test('a real launch failure after resume consumes the attempt at the unchanged prerequisite', { timeout: 30000 }, async t => {
  const x = await restingRequest(t, { brokenHolders: true }), { f } = x;
  const halted = await f.call('human', 'processing.halt');
  await expectQuietTicks(x, 'processing-halted');
  await f.call('human', 'processing.resume', { generation: halted.generation });
  await x.launcher.tick();
  assert.equal((await f.call('human', 'request.get', { request: f.request })).launch_refusal.code, 'launch-failed');
  assert.equal(x.chiefChecks(), 1, 'the chief launch was attempted once');
  const prerequisite = f.bus.requestLaunches('demo')[0].prerequisite;
  const haltedAgain = await f.call('human', 'processing.halt');
  await x.launcher.tick();
  assert.equal((await f.call('human', 'request.get', { request: f.request })).launch_refusal.code, 'launch-failed', 'the identical earlier halt receipt is deduplicated');
  await f.call('human', 'processing.resume', { generation: haltedAgain.generation });
  await x.replace(false);
  await x.launcher.tick();
  assert.equal((await f.call('human', 'request.get', { request: f.request })).launch_refusal.code, 'launches-disabled');
  await x.replace();
  assert.equal(f.bus.requestLaunches('demo')[0].prerequisite, prerequisite);
  const effects = f.journal.effects('demo', f.request).length;
  for (let i = 0; i < 5; i++) await x.launcher.tick();
  assert.equal(x.chiefChecks(), 1, 'a real failure does not retry at the same prerequisite');
  assert.equal(f.journal.effects('demo', f.request).length, effects);
  assert.equal(x.launcher.registry.sessions.length, 0);
});