// The S6 launcher inside a real bus host: real broker child, real runners
// under unshare, the fake adapter from packages/harness standing in for the
// harness. The PM launches a coder through its `launch` tool; every host
// refusal comes back over the launch socket and lands in the launch log.

import { test } from "node:test";
import assert from "node:assert/strict";
import { spawn, spawnSync } from "node:child_process";
import { once } from "node:events";
import { createInterface } from "node:readline";
import { connect } from "node:net";
import { chmodSync, existsSync, mkdirSync, readFileSync, readdirSync, statSync, writeFileSync } from "node:fs";
import { join } from "node:path";
import { fileURLToPath } from "node:url";
import { Client } from "../../bus/src/client.mjs";
import { launchLogFile, launchSocketPath, readSessions, Registry, sessionsFile, UNSHARE } from "../../seat/src/session.mjs";
import { bootConfig, loadSystem } from "../src/config.mjs";
import { startHost, startTimeOf } from "../src/host.mjs";
import { claudeVersion, createLauncher } from "../src/launcher.mjs";
import { fixture, tmp } from "./helpers.mjs";

const FAKE = fileURLToPath(new URL("../../harness/tests/fixtures/fake-adapter.mjs", import.meta.url));
const HOST = fileURLToPath(new URL("./fixtures/launch-host.mjs", import.meta.url));
const NAMESPACES = spawnSync(UNSHARE[0], [...UNSHARE.slice(1), "true"]).status === 0;
const skip = NAMESPACES ? false : "unprivileged user and PID namespaces are unavailable here";

// pm opus, coder and reviewer sonnet, one of each family; cto is listed but
// runs the system model, which is in no family.
function edit(doc) {
  doc.roles.coder.vars.model = "claude-sonnet-5-5";
  doc.roles.reviewer.vars = { harness: "pi", model: "claude-sonnet-5-5" };
  doc.launch = { by: "pm", instances: ["coder", "reviewer", "cto"], max: { opus: 1, sonnet: 1 } };
  return doc;
}

async function until(fn, ms = 15000) {
  const end = Date.now() + ms;
  while (Date.now() < end) {
    const v = await fn();
    if (v) return v;
    await new Promise((r) => setTimeout(r, 50));
  }
  throw new Error("timed out waiting");
}

// A config and data root with the fake adapter; setup() starts a host on it.
function prepare(t, more = (doc) => doc) {
  const root = tmp(t, "mosaic-launch-");
  const f = fixture(root, "acme", (doc) => more(edit(doc)));
  mkdirSync(f.dataRoot, { mode: 0o700 });
  const adapter = join(root, "adapter.sh");
  writeFileSync(adapter, `exec '${process.execPath}' '${FAKE}'\n`, { mode: 0o644 });
  return { root, f, adapter };
}

async function setup(t, more = (doc) => doc, { root, f, adapter } = prepare(t, more), options = {}) {
  const system = loadSystem({ env: f.env });
  const boot = bootConfig({ system, businessId: "acme", env: f.env });
  const logs = [];
  const host = await startHost({ boot, business: "acme", log: (l) => logs.push(l) });
  let launcher;
  try {
    launcher = createLauncher({
      host,
      system,
      env: f.env,
      adapters: { pi: adapter, "claude-code": adapter },
      sessionDefaults: { pollInterval: 100 },
      versions: { pi: "0.85.1", claude: null },
      log: (l) => logs.push(l),
      ...options,
    });
    await launcher.listen();
  } catch (e) {
    await host.close(1);
    throw e;
  }
  let closed = false;
  const close = async () => {
    if (closed) return host.done;
    closed = true;
    return host.close(0);
  };
  t.after(close);
  // A test caller: a cto launch bound to this process, claimed.
  const cto = await host.bindLaunch({ business: "acme", role: "cto", run: "cto-test", harness: "claude-code", pid: process.pid, startTime: startTimeOf(process.pid) });
  const client = new Client({ path: host.path, cap: cto.cap });
  await client.call("role.claim");
  return { root, f, host, launcher, cto, client, logs, close };
}

// One request on the launch socket, one reply line.
function ask(path, line) {
  return new Promise((resolve, reject) => {
    const s = connect(path);
    let buf = "";
    s.on("connect", () => s.write(line));
    s.on("data", (b) => (buf += b));
    s.on("end", () => resolve(JSON.parse(buf)));
    s.on("error", reject);
  });
}

const log = (f) => readFileSync(launchLogFile(f.dataRoot, "acme"), "utf8").trim().split("\n").map((l) => JSON.parse(l));

test("bus start --pm: the PM runs under its own PID namespace, registered, with a manifest", { skip, timeout: 30000 }, async (t) => {
  const { f, launcher, host, cto, close } = await setup(t);
  const sock = launchSocketPath(f.dataRoot);
  assert.equal(statSync(sock).mode & 0o777, 0o600);

  const pm = await launcher.launchPm();
  assert.deepEqual({ ...pm, run: undefined }, { run: undefined, instance: "pm", harness: "pi", model: "claude-opus-5-5" });
  assert.match(pm.run, /^r[0-9a-f]{12}$/);

  const [s] = readSessions(f.dataRoot);
  assert.equal(s.run, pm.run);
  assert.equal(s.live, true);
  assert.equal(s.family, "opus");
  assert.equal(s.launchedBy, "jason");
  assert.notEqual(s.pid, s.runnerPid, "the runner is the unshare process's child");
  assert.equal(statSync(sessionsFile(f.dataRoot)).mode & 0o777, 0o600);

  const runDir = s.runDir;
  assert.equal(statSync(runDir).mode & 0o777, 0o700);
  const session = JSON.parse(readFileSync(join(runDir, "session.json"), "utf8"));
  assert.equal(session.launchedBy, "jason");
  assert.equal(session.brokerSocket, host.path);
  assert.equal(session.launchSocket, sock);
  const manifest = JSON.parse(readFileSync(join(runDir, "bundle", "manifest.json"), "utf8"));
  assert.equal(manifest.piVersion, "0.85.1");
  assert.equal(manifest.claudeVersion, null);
  const [launched] = log(f).filter((e) => e.event === "launch");
  assert.equal(launched.run, pm.run);
  assert.equal(launched.manifest, join(runDir, "bundle", "manifest.json"));
  assert.equal(manifest.model, "claude-opus-5-5");

  // A second PM is refused before anything spawns.
  await assert.rejects(launcher.launchPm(), /PM launch refused: instance-running/);

  // The session's environment carries the run id and no capability; the
  // capability is nowhere on disk or on any session process's command line.
  await until(() => existsSync(join(runDir, "runner.log")) && readSessions(f.dataRoot)[0]?.live);
  const proc = (pid, what) => {
    try {
      return readFileSync(`/proc/${pid}/${what}`, "utf8");
    } catch {
      return "";
    }
  };
  const environ = proc(s.runnerPid, "environ").split("\0");
  assert.ok(environ.includes(`MOSAIC_RUN_ID=${pm.run}`));
  assert.ok(!environ.some((v) => v.startsWith("MOSAIC_CONFIG=")), "only allowlisted variables pass");
  const files = (dir) => readdirSync(dir, { recursive: true, withFileTypes: true }).filter((d) => d.isFile()).map((d) => join(d.parentPath, d.name));
  const capOf = cto.cap; // a known capability shape, to prove the scan finds nothing like it
  for (const file of files(f.dataRoot)) {
    const text = readFileSync(file, "latin1");
    assert.ok(!text.includes(capOf), `${file} holds no capability`);
  }
  for (const pid of [s.pid, s.runnerPid]) {
    assert.notEqual(proc(pid, "cmdline"), "");
    assert.ok(!/[0-9a-f]{64}/.test(proc(pid, "cmdline")), "no capability in argv");
    assert.ok(!/[0-9a-f]{64}/.test(proc(pid, "environ")), "no capability in the environment");
  }

  // Closing the host stops the session and ends the launch.
  assert.equal(await close(), 0);
  assert.equal(startTimeOf(s.runnerPid), null);
  assert.deepEqual(readSessions(f.dataRoot), []);
  assert.equal(existsSync(sock), false);
  const end = log(f).find((e) => e.event === "end");
  assert.equal(end.run, pm.run);
  assert.equal(end.reason, "stopped");
  assert.equal(end.exitCode, 0);
});

test("the PM launches a coder through its launch tool; the coder answers; refusals name their code", { skip, timeout: 30000 }, async (t) => {
  const { f, launcher, client, cto } = await setup(t);
  const sock = launchSocketPath(f.dataRoot);
  const pm = await launcher.launchPm();

  // cto asks pm to launch a coder; pm's turn calls the launch tool.
  const sent = await client.call("message.send", { to: "pm", class: "REQUEST", body: 'TOOL launch {"instance":"coder"}' });
  const reply = await until(async () => (await client.call("message.receive")).find((m) => m.in_reply_to === sent.id));
  assert.equal(reply.from_role, "pm");
  assert.equal(reply.class, "RESULT");
  const result = JSON.parse(reply.body);
  assert.equal(result.instance, "coder");
  assert.equal(result.model, "claude-sonnet-5-5");
  const coder = readSessions(f.dataRoot).find((s) => s.instance === "coder");
  assert.equal(coder.run, result.run);
  assert.equal(coder.launchedBy, "pm");
  assert.equal(coder.live, true);

  // The coder claimed its role and answers its own messages.
  const hello = await client.call("message.send", { to: "coder", class: "REQUEST", body: "hello" });
  const echo = await until(async () => (await client.call("message.receive")).find((m) => m.in_reply_to === hello.id));
  assert.equal(echo.from_role, "coder");
  assert.match(echo.body, /^echo: Message .* from cto, class REQUEST:\n\nhello/);

  // Host refusals, asked with the cto capability (steps 2 to 6 come before
  // the broker's role.launch check, which cto would fail).
  const refusals = [
    ["pm", "instance-not-listed"],
    ["nobody", "instance-not-listed"],
    ["coder", "instance-running"],
    ["reviewer", "capacity-full"],
    ["cto", "unknown-model-family"],
  ];
  for (const [instance, code] of refusals) {
    assert.deepEqual(await ask(sock, `${JSON.stringify({ cap: cto.cap, instance })}\n`), { ok: false, error: code }, instance);
  }
  // A listed, free instance with room still needs role.launch: cto lacks it.
  // Stop the coder first so sonnet has room again.
  const coderRun = coder.run;
  process.kill(coder.runnerPid, "SIGTERM");
  await until(() => !readSessions(f.dataRoot).some((s) => s.run === coderRun) && log(f).some((e) => e.event === "end" && e.run === coderRun));
  const denied = await ask(sock, `${JSON.stringify({ cap: cto.cap, instance: "reviewer" })}\n`);
  assert.equal(denied.ok, false);
  assert.notEqual(denied.error, "capacity-full");

  // Malformed requests.
  assert.deepEqual(await ask(sock, "not json\n"), { ok: false, error: "invalid-request" });
  assert.deepEqual(await ask(sock, `${JSON.stringify({ cap: "short", instance: "coder" })}\n`), { ok: false, error: "unauthenticated" });
  assert.deepEqual(await ask(sock, `${JSON.stringify({ cap: "0".repeat(64), instance: "coder" })}\n`), { ok: false, error: "unauthenticated" });
  assert.deepEqual(await ask(sock, `${JSON.stringify({ cap: cto.cap, instance: 7 })}\n`), { ok: false, error: "invalid-request" });

  // The launch log has the launches, every refusal, and the coder's end.
  const entries = log(f);
  assert.deepEqual(entries.filter((e) => e.event === "launch").map((e) => [e.instance, e.launchedBy]), [["pm", "jason"], ["coder", "pm"]]);
  const refused = entries.filter((e) => e.event === "refused").map((e) => [e.instance, e.code]);
  for (const r of refusals) assert.ok(refused.some(([i, c]) => i === r[0] && c === r[1]), `${r} is logged`);
  const end = entries.find((e) => e.event === "end" && e.run === coderRun);
  assert.equal(end.reason, "stopped");
  assert.ok(readSessions(f.dataRoot).some((s) => s.run === pm.run));
});

test("a runner that stops at once ends its launch with the runner's reason", { skip, timeout: 30000 }, async (t) => {
  // pm's tracker token has expired: the runner's founder-credential stop
  // exits 20 before role.claim, and the host records founder-credentials.
  const { f, launcher } = await setup(t, (doc) => {
    doc.roles.pm.credentials.vikunja.expires = "2000-01-01";
    return doc;
  });
  const pm = await launcher.launchPm();
  const end = await until(() => existsSync(launchLogFile(f.dataRoot, "acme")) && log(f).find((e) => e.event === "end" && e.run === pm.run));
  assert.equal(end.reason, "founder-credentials");
  assert.equal(end.exitCode, 20);
  assert.deepEqual(readSessions(f.dataRoot), []);
  assert.match(readFileSync(join(f.dataRoot, "launches", "acme", pm.run, "runner.log"), "utf8"), /vikunja/);
});

for (const exited of [false, true]) {
  test(`a host that died hard leaves its sessions to the next host, which ${exited ? "finds one already exited (23)" : "kills them"} and ends their runs so the role can be launched again`, { skip, timeout: 30000 }, async (t) => {
    const given = prepare(t);
    const { f } = given;
    const old = spawn(process.execPath, [HOST, given.adapter], { env: f.env, stdio: ["ignore", "pipe", "inherit"] });
    t.after(() => old.kill("SIGKILL"));
    const [line] = await once(createInterface({ input: old.stdout }), "line");
    const first = JSON.parse(line);
    const [left] = readSessions(f.dataRoot);
    assert.equal(left.run, first.run);
    // Killed before the claim, the runner would exit 23 on its own.
    const runnerLog = join(f.dataRoot, "launches", "acme", first.run, "runner.log");
    await until(() => existsSync(runnerLog) && readFileSync(runnerLog, "utf8").includes("claimed by run"));
    assert.equal(left.live, true);
    old.kill("SIGKILL");
    await once(old, "exit");
    // The broker child exits on the lost IPC channel; the session runs on.
    await until(() => !existsSync(join(f.dataRoot, "bus", "writer.lock")));
    assert.equal(startTimeOf(left.runnerPid), left.runnerStartTime);
    if (!exited) {
      // Stopped, the leftover neither polls the dead broker nor answers
      // SIGTERM: only the next host's SIGKILL ends it.
      for (const pid of [left.pid, left.runnerPid]) process.kill(pid, "SIGSTOP");
      t.after(() => {
        for (const [pid, start] of [[left.pid, left.startTime], [left.runnerPid, left.runnerStartTime]]) {
          if (startTimeOf(pid) === start) process.kill(pid, "SIGKILL");
        }
      });
    }
    if (exited) {
      await until(() => startTimeOf(left.pid) === null);
      assert.match(readFileSync(runnerLog, "utf8"), /broker unreachable after 30 tries[\s\S]*exiting 23/);
      assert.equal(readSessions(f.dataRoot)[0].live, false);
    }

    const s = await setup(t, undefined, given);
    assert.equal(startTimeOf(left.pid), null, "the old session process is gone");
    assert.equal(startTimeOf(left.runnerPid), null, "and its runner with it");
    assert.ok(s.logs.includes(`run ${first.run} (pm) from an earlier host ended: host-lost`), s.logs.join("\n"));
    const ended = (await s.client.call("trail", { subject: first.run })).filter((r) => r.kind === "session.ended");
    assert.equal(ended.length, 1);
    assert.deepEqual(ended[0].body, { reason: "host-lost", exitCode: null, released: true });
    const log = readFileSync(launchLogFile(f.dataRoot, "acme"), "utf8").trim().split("\n").map((l) => JSON.parse(l));
    assert.deepEqual(log.filter((e) => e.event === "end").map((e) => [e.run, e.reason]), [[first.run, "host-lost"]]);
    assert.deepEqual(readSessions(f.dataRoot), []);

    // Without the end, the old run would still hold pm and this runner would exit 21.
    const again = await s.launcher.launchPm();
    await until(async () => (await s.client.call("agents")).some((a) => a.role === "pm" && a.holder_run === again.run));
    await s.client.call("role.release"); // the test's own cto claim, so the next host's can claim again
    assert.equal(await s.close(), 0);

    // Died after the end but before sessions.json changed: the run stays ended.
    const { live, ...entry } = left;
    new Registry(f.dataRoot).add(entry);
    const third = await setup(t, undefined, given);
    assert.ok(third.logs.includes(`run ${first.run} (pm) from an earlier host was already ended`), third.logs.join("\n"));
    assert.equal((await third.client.call("trail", { subject: first.run })).filter((r) => r.kind === "session.ended").length, 1);
    assert.equal(await third.close(), 0);
  });
}

test("a malformed sessions.json refuses the host start with exit 3 and stays as it was", { timeout: 30000 }, async (t) => {
  const given = prepare(t);
  const file = sessionsFile(given.f.dataRoot);
  mkdirSync(join(given.f.dataRoot, "bus-host"), { mode: 0o700 });
  writeFileSync(file, "{", { mode: 0o600 });
  await assert.rejects(setup(t, undefined, given), (e) => e.exitCode === 3 && /sessions file is unreadable.*move the file aside/.test(e.message));
  assert.equal(readFileSync(file, "utf8"), "{");
  assert.equal(existsSync(join(given.f.dataRoot, "bus", "writer.lock")), false, "the host closed");
});

const within = (p, ms, what) => Promise.race([p, new Promise((_, reject) => setTimeout(() => reject(new Error(`${what} took over ${ms} ms`)), ms).unref())]);

test("an over-long launch request is refused at once, not at the 10 s idle timeout", { timeout: 30000 }, async (t) => {
  const { f } = await setup(t);
  const s = connect(launchSocketPath(f.dataRoot));
  t.after(() => s.destroy());
  let buf = "";
  s.on("data", (b) => (buf += b));
  s.write("x".repeat(5000));
  await within(once(s, "end"), 3000, "the refusal");
  assert.deepEqual(JSON.parse(buf), { ok: false, error: "invalid-request" });
});

test("a launch client that never closes its side doesn't hold the host's close", { timeout: 30000 }, async (t) => {
  const { f, close } = await setup(t);
  const path = launchSocketPath(f.dataRoot);
  // One got its reply and never ends; one never sends anything.
  const answered = connect({ path, allowHalfOpen: true });
  const silent = connect({ path, allowHalfOpen: true });
  const connected = once(silent, "connect");
  const gone = Promise.all([once(answered, "close"), once(silent, "close")]);
  const drop = () => {
    answered.destroy();
    silent.destroy();
  };
  t.after(drop);
  let buf = "";
  answered.on("data", (b) => (buf += b));
  answered.write("not json\n");
  await once(answered, "end");
  assert.deepEqual(JSON.parse(buf), { ok: false, error: "invalid-request" });
  await within(connected, 2000, "connect");
  // On a failure the clients go, so the host can still close and the test
  // fails instead of hanging.
  const code = await within(close(), 5000, "close").catch((e) => {
    drop();
    throw e;
  });
  assert.equal(code, 0);
  assert.equal(existsSync(path), false);
  answered.end();
  silent.end();
  await within(gone, 2000, "the clients' close");
});

test("a runner that ignores SIGTERM is killed when the host closes", { skip, timeout: 60000 }, async (t) => {
  const { f, launcher, close } = await setup(t, undefined, undefined, { stopTimeoutMs: 1000 });
  const pm = await launcher.launchPm();
  const runnerLog = join(f.dataRoot, "launches", "acme", pm.run, "runner.log");
  await until(() => existsSync(runnerLog) && readFileSync(runnerLog, "utf8").includes("claimed by run"));
  const [s] = readSessions(f.dataRoot);
  // Stopped from outside its namespace, it can't act on SIGTERM.
  process.kill(s.runnerPid, "SIGSTOP");
  const kill = () => {
    if (startTimeOf(s.runnerPid) === s.runnerStartTime) process.kill(s.runnerPid, "SIGKILL");
  };
  t.after(kill);
  const started = Date.now();
  const code = await within(close(), 15000, "close").catch((e) => {
    kill();
    throw e;
  });
  assert.equal(code, 0);
  assert.ok(Date.now() - started >= 1000, "it waited for the stop timeout");
  assert.equal(startTimeOf(s.pid), null);
  assert.equal(startTimeOf(s.runnerPid), null);
  assert.deepEqual(readSessions(f.dataRoot), []);
  const ends = log(f).filter((e) => e.event === "end" && e.run === pm.run);
  assert.equal(ends.length, 1);
  assert.equal(ends[0].signal, "SIGKILL");
});

// Row 68: the manifest's claudeVersion asks the binary sessions run.
test("claudeVersion asks MOSAIC_CLAUDE_BIN when set, else claude on PATH", (t) => {
  const dir = tmp(t);
  mkdirSync(join(dir, "bin"));
  const stand = (p, v) => {
    writeFileSync(p, `#!/bin/sh\necho "${v} (Claude Code)"\n`);
    chmodSync(p, 0o755);
    return p;
  };
  stand(join(dir, "bin", "claude"), "1.0.0");
  const pinned = stand(join(dir, "pinned"), "9.9.9");
  const PATH = join(dir, "bin");
  assert.equal(claudeVersion({ PATH, MOSAIC_CLAUDE_BIN: pinned }), "9.9.9 (Claude Code)");
  assert.equal(claudeVersion({ PATH }), "1.0.0 (Claude Code)");
  assert.equal(claudeVersion({ PATH: join(dir, "none") }), null);
});