// The runner against a real broker on a real socket, with a fake adapter.
// The broker runs in this process; the runner is a child, as the host
// starts it (minus the PID namespace, which packages/seat tests).

import { test } from "node:test";
import assert from "node:assert/strict";
import { spawn, spawnSync } from "node:child_process";
import { once } from "node:events";
import { existsSync, mkdirSync, readFileSync, rmSync, statSync, writeFileSync } from "node:fs";
import { open } from "node:fs/promises";
import { createServer } from "node:net";
import { join } from "node:path";
import { Store } from "../../bus/src/store.mjs";
import { Broker } from "../../bus/src/broker.mjs";
import { serve } from "../../bus/src/server.mjs";
import { EXIT, founderCheck, turnRequest } from "../src/runner.mjs";
import { scratch } from "./helpers.mjs";

const RUNNER = new URL("../src/runner.mjs", import.meta.url).pathname;
const FAKE = new URL("./fixtures/fake-adapter.mjs", import.meta.url).pathname;
const businesses = {
  demo: {
    id: "demo",
    human: "jason",
    arbiters: { technical: "pm", delivery: "pm" },
    launch: { by: "pm", instances: ["coder"], max: { opus: 2, sonnet: 2 } },
    roles: {
      pm: { authority: { withinRole: ["message.send", "role.launch"], crossRole: [] } },
      coder: { authority: { withinRole: ["message.send"], crossRole: [] } },
    },
  },
};

test("founderCheck: founder variables, then a needed service without a usable token", () => {
  const policy = { needs: ["gitea"], credentials: [{ service: "gitea", state: "valid" }] };
  assert.equal(founderCheck(policy, {}), null);
  assert.equal(founderCheck({ needs: [] }, { PATH: "/bin" }), null);
  assert.match(founderCheck(policy, { GITEA_TOKEN: "x", SSH_AUTH_SOCK: "/s" }), /GITEA_TOKEN, SSH_AUTH_SOCK/);
  assert.match(founderCheck(policy, { MOSAIC_GITEA_CREDENTIAL_FILE: "" }), /MOSAIC_GITEA_CREDENTIAL_FILE/);
  for (const state of ["expiring", "rotation-due"]) assert.equal(founderCheck({ needs: ["x"], credentials: [{ service: "x", state }] }, {}), null);
  assert.match(founderCheck({ needs: ["vikunja"], credentials: [{ service: "vikunja", state: "expired" }] }, {}), /no usable role token for vikunja/);
  assert.match(founderCheck({ needs: ["gitea", "vikunja"], credentials: [{ service: "gitea", state: "valid" }] }, {}), /vikunja/);
});

test("turnRequest names the sender, class, reply and decision", () => {
  assert.equal(turnRequest({ id: "m1", from_role: "pm", class: "ASSIGNMENT", body: "do it" }), "Message m1 from pm, class ASSIGNMENT:\n\ndo it");
  assert.equal(
    turnRequest({ id: "m2", from_role: "cto", class: "RESULT", in_reply_to: "m1", decision: "d1", body: "b" }),
    "Message m2 from cto, class RESULT, in reply to m1, citing decision d1:\n\nb",
  );
});

async function setup(t, { session = {}, policy = {}, tools } = {}) {
  const dir = scratch(t);
  const store = new Store(dir);
  const broker = new Broker({ store, businesses });
  const brokerSocket = join(dir, "bus.sock");
  const server = await serve({ broker, path: brokerSocket });
  t.after(async () => {
    await server.close().catch(() => {});
    store.close();
  });
  const runDir = join(dir, "run");
  const bundle = join(runDir, "bundle");
  mkdirSync(bundle, { recursive: true, mode: 0o700 });
  const adapter = join(dir, "adapter.sh");
  writeFileSync(adapter, `exec '${process.execPath}' '${FAKE}'\n`);
  writeFileSync(join(bundle, "prompt.md"), "# test\n");
  writeFileSync(join(bundle, "policy.json"), JSON.stringify({ harness: "pi", workspace: join(dir, "ws"), tools: [], typed: ["send_message", "launch"], network: "none", needs: [], credentials: [], ...policy }));
  writeFileSync(join(bundle, "tools.json"), JSON.stringify(tools ?? [
    { name: "send_message", verb: "message.send", action: "message.send" },
    { name: "launch", verb: "launch", action: "role.launch" },
  ]));
  mkdirSync(join(dir, "ws"));
  const launches = [];
  const launchSocket = join(dir, "launch.sock");
  const launchServer = createServer((s) => {
    s.once("data", (b) => {
      const r = JSON.parse(b);
      launches.push(r);
      s.end(JSON.stringify(r.instance === "coder" ? { ok: false, error: "capacity-full" } : { ok: true, result: { run: "new-run" } }) + "\n");
    });
  });
  await new Promise((r) => launchServer.listen(launchSocket, r));
  t.after(() => launchServer.close());
  writeFileSync(join(runDir, "session.json"), JSON.stringify({
    business: "demo",
    instance: "coder",
    run: "coder-run",
    launchedBy: "pm",
    harness: "pi",
    provider: "fake",
    model: "fake-model",
    brokerSocket,
    launchSocket,
    workspace: join(dir, "ws"),
    sessionDir: join(runDir, "session"),
    adapter,
    bundle: { prompt: join(bundle, "prompt.md"), policy: join(bundle, "policy.json"), tools: join(bundle, "tools.json") },
    toolSocket: join(runDir, "tools.sock"),
    turnMarker: join(runDir, "turn.done"),
    pollInterval: 50,
    turnTimeout: 20,
    ...session,
  }));
  const cap = broker.bindLaunch({ business: "demo", role: "coder", run: "coder-run", harness: "pi" });
  const pm = broker.bindLaunch({ business: "demo", role: "pm", run: "pm-run", harness: "pi" });
  const call = (c, verb, args = {}) => broker.request(c, { verb, args });
  call(pm, "role.claim");
  return { dir, runDir, store, broker, server, cap, pm, call, launches, pids: join(dir, "pids") };
}

// A runner that hasn't exited after a minute is killed, so a test that
// expected an exit fails on the code instead of hanging.
function startRunner(runDir, cap, env = {}, input = cap === null ? "" : JSON.stringify({ cap }) + "\n") {
  const child = spawn(process.execPath, [RUNNER, runDir], {
    stdio: ["pipe", "ignore", "pipe"],
    env: { PATH: process.env.PATH, ...env },
  });
  let stderr = "";
  child.stderr.on("data", (b) => (stderr += b));
  child.stdin.on("error", () => {});
  child.stdin.end(input);
  const clock = setTimeout(() => child.kill("SIGKILL"), 60_000);
  const exited = once(child, "exit").then(([code, signal]) => {
    clearTimeout(clock);
    return { code, signal, stderr };
  });
  return { child, exited, stderr: () => stderr };
}

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

const replies = (ctx) => () => {
  const got = ctx.call(ctx.pm, "message.receive");
  ctx.inbox.push(...got);
  return ctx.inbox.length ? ctx.inbox : null;
};

test("a message becomes a turn, the answer goes back as a RESULT, SIGTERM releases and exits 0", async (t) => {
  const ctx = await setup(t);
  ctx.inbox = [];
  const r = startRunner(ctx.runDir, ctx.cap);
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  const sent = ctx.call(ctx.pm, "message.send", { to: "coder", body: "hello", class: "REQUEST" });
  const [reply] = await until(replies(ctx));
  assert.equal(reply.class, "RESULT");
  assert.equal(reply.from_role, "coder");
  assert.equal(reply.from_run, "coder-run");
  assert.equal(reply.in_reply_to, sent.id);
  assert.equal(reply.body, `echo: Message ${sent.id} from pm, class REQUEST:\n\nhello`);
  await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", sent.id));
  const err = join(ctx.runDir, "turns", "0001.stderr");
  assert.equal(statSync(err).mode & 0o777, 0o600);
  assert.equal(readFileSync(err, "utf8"), "fake adapter: fake-model\n");
  assert.equal(statSync(join(ctx.runDir, "tools.sock")).mode & 0o777, 0o600);
  r.child.kill("SIGTERM");
  const out = await r.exited;
  assert.equal(out.code, EXIT.stopped, out.stderr);
  assert.match(out.stderr, /runner: demo\/coder claimed by run coder-run/);
  assert.equal(ctx.store.get("SELECT op FROM role_claims WHERE role='coder' ORDER BY seq DESC LIMIT 1").op, "release");
  assert.ok(!existsSync(join(ctx.runDir, "tools.sock")));
});

test("a SIGTERM before the claim stops the runner with exit 0 and no claim", async (t) => {
  const ctx = await setup(t);
  // session.json becomes a FIFO: main opens it right after its handlers go
  // in, so the open on this side returns once a SIGTERM would be caught.
  const file = join(ctx.runDir, "session.json");
  const body = readFileSync(file);
  rmSync(file);
  assert.equal(spawnSync("mkfifo", ["-m", "600", file]).status, 0);
  const child = spawn(process.execPath, [RUNNER, ctx.runDir], { stdio: ["pipe", "ignore", "pipe"], env: { PATH: process.env.PATH } });
  t.after(() => child.kill("SIGKILL"));
  let stderr = "";
  child.stderr.on("data", (b) => (stderr += b));
  const exited = once(child, "exit");
  const fifo = await open(file, "w");
  child.kill("SIGTERM");
  await fifo.write(body);
  await fifo.close();
  child.stdin.end(JSON.stringify({ cap: ctx.cap }) + "\n");
  const [code, signal] = await exited;
  assert.deepEqual([code, signal], [EXIT.stopped, null], stderr);
  assert.match(stderr, /runner: stopped before claim/);
  assert.equal(ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder'"), undefined);
});

test("typed tools carry the runner's capability; launch goes to the host's launch socket", async (t) => {
  const ctx = await setup(t);
  ctx.inbox = [];
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  ctx.call(ctx.pm, "message.send", { to: "coder", body: 'TOOL send_message {"to":"pm","body":"side note","class":"INFO"}' });
  await until(() => replies(ctx)() && ctx.inbox.length >= 2);
  const [info, result] = ctx.inbox;
  assert.equal(info.class, "INFO");
  assert.equal(info.body, "side note");
  assert.equal(info.from_run, "coder-run");
  assert.equal(JSON.parse(result.body).id, info.id);
  ctx.inbox.length = 0;
  ctx.call(ctx.pm, "message.send", { to: "coder", body: 'TOOL launch {"instance":"coder"}' });
  ctx.call(ctx.pm, "message.send", { to: "coder", body: 'TOOL launch {"instance":"reviewer"}' });
  ctx.call(ctx.pm, "message.send", { to: "coder", body: 'TOOL resolve_decision {"id":"x"}' });
  ctx.call(ctx.pm, "message.send", { to: "coder", body: 'TOOL send_message {"to":"nobody","body":"x"}' });
  await until(() => replies(ctx)() && ctx.inbox.length >= 4);
  assert.deepEqual(ctx.inbox.map((m) => m.body), ["refused: capacity-full", '{"run":"new-run"}', "refused: unknown-tool", "refused: unknown-role"]);
  assert.deepEqual(ctx.launches, [{ cap: ctx.cap, instance: "coder" }, { cap: ctx.cap, instance: "reviewer" }]);
});

test("a RESULT gets no automatic reply; failed turns reply with the reason", async (t) => {
  const ctx = await setup(t, { session: { turnTimeout: 1 } });
  ctx.inbox = [];
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  const first = ctx.call(ctx.pm, "message.send", { to: "coder", body: "hi" });
  const res = ctx.call(ctx.pm, "message.send", { to: "coder", body: "thanks", class: "RESULT", in_reply_to: first.id });
  ctx.call(ctx.pm, "message.send", { to: "coder", body: "FAIL" });
  ctx.call(ctx.pm, "message.send", { to: "coder", body: "NOMARKER" });
  ctx.call(ctx.pm, "message.send", { to: "coder", body: "SLEEP" });
  await until(() => replies(ctx)() && ctx.inbox.length >= 4);
  assert.deepEqual(ctx.inbox.map((m) => m.body.replace(/^echo: .*/s, "echo")), [
    "echo",
    "The turn failed: adapter exited 3",
    "The turn failed: pi exited before the turn ended (no agent_end)",
    "The turn failed: turn passed its 1 s wall clock",
  ]);
  await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", res.id));
});

test("SIGTERM during a turn kills the turn's process group and still exits 0", async (t) => {
  // A long wall clock, so only the stop path can end the turn in time.
  const ctx = await setup(t, { session: { turnTimeout: 600 } });
  const r = startRunner(ctx.runDir, ctx.cap, { FAKE_PIDS: ctx.pids });
  t.after(() => r.child.kill("SIGKILL"));
  ctx.call(ctx.pm, "message.send", { to: "coder", body: "SLEEP" });
  const pid = Number(await until(() => existsSync(ctx.pids) && readFileSync(ctx.pids, "utf8").trim()));
  const started = Date.now();
  r.child.kill("SIGTERM");
  const out = await Promise.race([r.exited, new Promise((res) => setTimeout(() => res({ code: "still running" }), 8000))]);
  assert.equal(out.code, EXIT.stopped, out.stderr);
  assert.ok(Date.now() - started < 8000);
  await until(() => {
    try {
      process.kill(pid, 0);
      return false;
    } catch {
      return true;
    }
  }, 8000);
});

test("founder credentials stop before the claim (20)", async (t) => {
  const ctx = await setup(t);
  const out = await startRunner(ctx.runDir, ctx.cap, { GH_TOKEN: "x" }).exited;
  assert.equal(out.code, EXIT.founder);
  assert.match(out.stderr, /REQ-CRED-2.*GH_TOKEN/);
  assert.equal(ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder'"), undefined);
  const ctx2 = await setup(t, { policy: { needs: ["gitea"], credentials: [{ service: "gitea", state: "expired" }] } });
  const out2 = await startRunner(ctx2.runDir, ctx2.cap).exited;
  assert.equal(out2.code, EXIT.founder);
  assert.match(out2.stderr, /no usable role token for gitea/);
});

test("a refused claim exits 21; an ended run's capability exits 22", async (t) => {
  const ctx = await setup(t);
  const other = ctx.broker.bindLaunch({ business: "demo", role: "coder", run: "other-run", harness: "pi" });
  ctx.call(other, "role.claim");
  const out = await startRunner(ctx.runDir, ctx.cap).exited;
  assert.equal(out.code, EXIT.claim, out.stderr);
  assert.match(out.stderr, /role.claim refused: role-already-held/);
  const ctx2 = await setup(t);
  ctx2.broker.endLaunch({ business: "demo", run: "coder-run", reason: "stopped", exitCode: 0 });
  assert.equal((await startRunner(ctx2.runDir, ctx2.cap).exited).code, EXIT.session);
});

test("the launch ending under a running session exits 22", async (t) => {
  const ctx = await setup(t);
  const r = startRunner(ctx.runDir, ctx.cap);
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  ctx.broker.endLaunch({ business: "demo", run: "coder-run", reason: "killed", exitCode: null });
  const out = await r.exited;
  assert.equal(out.code, EXIT.session, out.stderr);
  assert.match(out.stderr, /session stopped working: unauthenticated/);
});

test("a broker that stays unreachable exits 23 after brokerRetries polls", async (t) => {
  const ctx = await setup(t, { session: { brokerRetries: 3 } });
  const r = startRunner(ctx.runDir, ctx.cap);
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  await ctx.server.close();
  const out = await r.exited;
  assert.equal(out.code, EXIT.broker, out.stderr);
  assert.match(out.stderr, /broker unreachable after 3 tries/);
});

test("a broker that is down at the claim exits 23, not 21", async (t) => {
  const ctx = await setup(t);
  await ctx.server.close();
  const out = await startRunner(ctx.runDir, ctx.cap).exited;
  assert.equal(out.code, EXIT.broker, out.stderr);
  assert.match(out.stderr, /role.claim got no answer from the broker/);
});

test("no capability, or a malformed one, on stdin exits 2", async (t) => {
  const ctx = await setup(t);
  assert.equal((await startRunner(ctx.runDir, null).exited).code, EXIT.usage);
  assert.equal((await startRunner(ctx.runDir, "not-hex").exited).code, EXIT.usage);
  assert.equal((await startRunner(join(ctx.dir, "nope"), ctx.cap).exited).code, EXIT.usage);
  // A valid capability inside an over-long line: the 4096-byte cap refuses it.
  const long = await startRunner(ctx.runDir, ctx.cap, {}, JSON.stringify({ cap: ctx.cap, pad: "x".repeat(5000) }) + "\n").exited;
  assert.equal(long.code, EXIT.usage, long.stderr);
  assert.match(long.stderr, /stdin too large/);
  assert.equal(ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder'"), undefined);
});

test("a missing or malformed policy exits 2 before the claim", async (t) => {
  const ctx = await setup(t);
  const policy = join(ctx.runDir, "bundle", "policy.json");
  writeFileSync(policy, "{");
  const bad = await startRunner(ctx.runDir, ctx.cap).exited;
  assert.equal(bad.code, EXIT.usage, bad.stderr);
  assert.match(bad.stderr, /^runner: /m);
  rmSync(policy);
  const missing = await startRunner(ctx.runDir, ctx.cap).exited;
  assert.equal(missing.code, EXIT.usage, missing.stderr);
  assert.match(missing.stderr, /runner: ENOENT/);
  assert.equal(ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder'"), undefined);
});