import { requestHostOps } from './request-host-ops.mjs';
// Clear context (row 271, 264d, #1776, T4): the launcher half of D7 and D8. A clear of a running
// session stops its drained run once, retires it on the run's own end, and starts the successor
// through the same check()/start() as any session; a held start waits for capacity; an earlier
// host's live run of the cleared session holds the clear until it exits; and a session runs in its
// own pid namespace (N6); a bus that can't answer holds a clear where it is (T6). Real runners in
// the bwrap sandbox, with clear-adapter.mjs as the turn.
// The bus half is packages/bus/tests/session-clear.test.mjs.
//
// The host is session-launch.test.mjs's, with the five clear ops the broker child forwards.

import { test } from "node:test";
import assert from "node:assert/strict";
import { spawn, spawnSync } from "node:child_process";
import { existsSync, readFileSync, writeFileSync } from "node:fs";
import { join } from "node:path";
import { fileURLToPath } from "node:url";
import { launchDoc } from "../../organization/tests/helpers.mjs";
import { BusError } from "../../bus/src/broker.mjs";
import { startBroker } from "../../bus/src/runtime.mjs";
import { gitEnv } from "../../seat/src/checkout.mjs";
import { startTimeOf } from "../../seat/src/proc.mjs";
import { BWRAP, sandboxProblem } from "../../seat/src/sandbox.mjs";
import { launchLogFile, readSessions, Registry } from "../../seat/src/session.mjs";
import { bootConfig, loadSystem } from "../src/config.mjs";
import { createLauncher } from "../src/launcher.mjs";
import { fakeHarnesses, fixture, tmp, withoutHostCredentials } from "./helpers.mjs";

const ADAPTER = fileURLToPath(new URL("./fixtures/clear-adapter.mjs", import.meta.url));
const skip = sandboxProblem() ?? false;

async function until(fn, ms = 20000) {
  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 mutant's kill is an assertion failure, not a timeout (sage-2707): the waits return what they
// last saw and the assertions after them decide.
async function settle(fn, ms = 20000) {
  const end = Date.now() + ms;
  let v;
  while (Date.now() < end) {
    v = await fn();
    if (v) return v;
    await new Promise((r) => setTimeout(r, 50));
  }
  return v;
}

function gitIn(cwd, ...args) {
  const r = spawnSync("git", ["-c", "user.name=test", "-c", "user.email=test@example.invalid", "-C", cwd, ...args], { env: gitEnv(), encoding: "utf8" });
  assert.equal(r.status, 0, r.stderr);
}

// The assistant is the thread instance; sonnet is its model's limit.
function prepare(t, { sonnet = 3, idle = 30 } = {}) {
  const root = tmp(t, "mosaic-clear-");
  const f = fixture(root, "acme", (doc) => {
    doc.roles.chief.vars = { harness: "pi", model: "claude-opus-5-5", thinking: "high" };
    doc.roles.assistant = { definition: "assistant", vars: { harness: "pi", model: "claude-sonnet-5-5", thinking: "high" } };
    doc.launch = launchDoc("chief", ["coder", "assistant"], { opus: 1, sonnet });
    doc.launch.threadIdleSeconds = idle;
    return doc;
  });
  withoutHostCredentials(f.env, { piKey: true });
  const project = f.doc.projects.stack.root;
  gitIn(project, "init", "-q", "-b", "main");
  writeFileSync(join(project, "file.txt"), "one\n");
  gitIn(project, "add", "file.txt");
  gitIn(project, "commit", "-q", "-m", "one");
  const adapter = join(root, "adapter.sh");
  writeFileSync(adapter, `exec '${process.execPath}' '${ADAPTER}'\n`, { mode: 0o644 });
  return { root, f, adapter };
}

async function start(t, { f, adapter }, { tickMs = 100, stopTimeoutMs } = {}) {
  const system = loadSystem({ env: f.env });
  const { trackers, ...boot } = bootConfig({ system, businessId: "acme", env: f.env });
  assert.equal(trackers, undefined);
  const runtime = await startBroker(boot);
  const ops = {
    ...requestHostOps(runtime),
    identity: (m) => runtime.identity(m.cap),
    authorizeLaunch: (m) => runtime.authorizeLaunch(m.cap, m.instance),
    refuse: (m) => runtime.refuse(m.cap, m.code, m.subject),
    endLaunch: (m) => runtime.endLaunch(m.record),
    recoverLaunch: (m) => runtime.recoverLaunch(m.record),
    credentialStatus: (m) => runtime.credentialStatus(m.business, m.role),
    threadsState: (m) => runtime.threadsState(m.business),
    threadIdle: (m) => runtime.threadIdle(m.record),
    mayLaunch: (m) => runtime.mayLaunch(m),
    landHeld: (m) => runtime.landHeld(m.record),
    landHolders: (m) => runtime.landHolders(m.business),
    landRelease: (m) => runtime.landRelease(m.record),
    sessionClears: (m) => runtime.sessionClears(m.business),
    sessionClearExited: (m) => runtime.sessionClearExited(m.record),
    sessionClearUncertain: (m) => runtime.sessionClearUncertain(m.record),
    sessionClearRetire: (m) => runtime.sessionClearRetire(m.record),
    sessionClearStart: (m) => runtime.sessionClearStart(m.record),
  };
  const refusing = (what, fn) => async (m) => {
    try {
      return fn(m);
    } catch (e) {
      const code = e instanceof BusError ? e.code : `${what}-refused`;
      throw Object.assign(new Error(`${what} refused: ${code}`), { code });
    }
  };
  const hooks = [];
  let closing = null;
  const host = {
    business: "acme",
    path: runtime.path,
    dataRoot: f.dataRoot,
    op: refusing("launch-op", (m) => {
      if (!Object.hasOwn(ops, m.op)) throw new BusError("invalid-host-request");
      return ops[m.op](m);
    }),
    bindLaunch: refusing("launch bind", (r) => runtime.bindLaunch(r)),
    beforeClose: (fn) => void hooks.push(fn),
  };
  const logs = [];
  const launcher = createLauncher({
    host,
    system,
    env: f.env,
    makeLoader: await fakeHarnesses(t, { pi: adapter, "claude-code": adapter }),
    sessionDefaults: { pollInterval: 100 },
    harnessVersion: "0.85.1",
    readOnly: [ADAPTER],
    threadTickMs: tickMs,
    ...(stopTimeoutMs ? { stopTimeoutMs } : {}),
    log: (l) => logs.push(l),
  });
  const close = () =>
    (closing ??= (async () => {
      for (const fn of hooks) await fn(0);
      await runtime.close();
    })());
  t.after(close);
  await launcher.listen();
  const human = runtime.broker.bindHuman({ business: "acme", human: f.doc.human, via: "cli", outsideAgent: true });
  const call = (verb, args = {}) => runtime.broker.request(human, { verb, args });
  return { host, runtime, launcher, call, logs, close };
}

const launchLog = (f) =>
  existsSync(launchLogFile(f.dataRoot, "acme"))
    ? readFileSync(launchLogFile(f.dataRoot, "acme"), "utf8").trim().split("\n").map((l) => JSON.parse(l))
    : [];
const events = (f, event) => launchLog(f).filter((e) => e.event === event);
const sessionWs = (f, session) => join(f.dataRoot, "session-workspaces", "acme", "assistant", session);
const historyOf = (f, thread, session) => {
  const p = join(f.dataRoot, "threads", "acme", thread, "sessions", session, "session", "history");
  return existsSync(p) ? readFileSync(p, "utf8") : null;
};

function send(call, thread, session, body) {
  const sent = call("thread.send", { thread, ...(session ? { session } : {}), body });
  const reply = async () => {
    const r = await until(() => call("thread.messages", { thread, ...(session ? { session } : {}) }).find((m) => m.in_reply_to === sent.id && m.class === "RESULT"));
    assert.equal(r.from_role, "assistant");
    return JSON.parse(r.body.match(/THREAD (\{.*\})/)[1]);
  };
  return { sent, reply };
}

function thread(call) {
  const project = call("project.create", { name: "stack work", repo: "stack" }).id;
  const A = call("thread.create", { project, name: "alpha" }).id;
  const [{ id: D }] = call("session.list", { thread: A });
  return { project, A, D };
}

test("REQ-THREAD-2 T4 D7 D8 N6: a clear of a running session stops its drained run once, retires it on the run's end, and starts the successor fresh in its own namespace", { skip, timeout: 90000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  const host = await start(t, prepared);
  const { A, D } = thread(host.call);
  const first = await send(host.call, A, D, "before the clear").reply();
  assert.deepEqual([first.before, first.cwd], [[], sessionWs(f, D)]);
  // N6: the run sees its own pid namespace. Pid 1 there is the sandboxed command, not bwrap, and
  // the host's processes aren't listed.
  assert.notEqual(first.pid1[0], BWRAP);
  assert.ok(first.pids.length < 20, JSON.stringify(first.pids));
  assert.equal(first.pids.includes(1), true);
  const [{ run }] = events(f, "launch");

  const op = host.call("session.clear", { thread: A, session: D, revision: 0, confirm: true, op: "clear-1" });
  const S = op.successor;
  const done = await settle(() => {
    const e = host.call("session.operation", { thread: A, op: "clear-1" });
    return e.start === "started" && e;
  }, 30000);
  assert.deepEqual([done.state, done.start, done.refusal], ["retired", "started", null]);
  // The drained run got one stop, and ended stopped; the clear retired it after that end.
  const stops = events(f, "session-clear-stop");
  assert.deepEqual(stops.map((e) => [e.session, e.run, e.op]), [[D, run, "clear-1"]]);
  assert.deepEqual(events(f, "session-clear-uncertain"), []);
  const kinds = launchLog(f).map((e) => [e.event, e.session ?? null]).filter(([e]) => ["session-clear-stop", "end", "session-clear-retired", "launch"].includes(e));
  assert.deepEqual(kinds, [
    ["launch", D],
    ["session-clear-stop", D],
    ["end", D],
    ["session-clear-retired", D],
    ["launch", S],
  ]);
  assert.deepEqual(events(f, "session-clear-retired").map((e) => [e.successor, e.op]), [[S, "clear-1"]]);
  // The successor is the default now; a thread-only send reaches it, with no history and its own
  // workspace. The old session's store stays as it was.
  const after = await send(host.call, A, null, "after the clear").reply();
  assert.deepEqual([after.body, after.before, after.cwd], ["after the clear", [], sessionWs(f, S)]);
  assert.equal(historyOf(f, A, D), "before the clear\n");
  assert.equal(historyOf(f, A, S), "after the clear\n");
  assert.deepEqual(host.call("session.list", { thread: A }).map((r) => [r.id, r.state, r.default]), [
    [D, "retired", false],
    [S, "active", true],
  ]);
  assert.deepEqual(readSessions(f.dataRoot).map((s) => s.session), [S]);
});

test("REQ-THREAD-2 T4 D8: a successor start refused capacity-full is held, reported and logged once, and starts when capacity frees", { skip, timeout: 90000 }, async (t) => {
  const prepared = prepare(t, { sonnet: 1, idle: 4 });
  const { f } = prepared;
  const host = await start(t, prepared);
  const { A, D } = thread(host.call);
  const N = host.call("session.create", { thread: A, name: "side", op: "op-side" }).id;
  // The side session takes the one sonnet slot; the default never ran, so its clear has no run.
  await send(host.call, A, N, "fill the slot").reply();
  const S = host.call("session.clear", { thread: A, session: D, revision: 0, confirm: true, op: "clear-1" }).successor;
  const held = await settle(() => {
    const e = host.call("session.operation", { thread: A, op: "clear-1" });
    return e.start === "held" && e;
  });
  assert.deepEqual([held.state, held.start, held.refusal], ["retired", "held", null]);
  assert.deepEqual(events(f, "session-clear-stop"), [], "no run to stop");
  // The side session idles out, and the held start goes on the next tick after.
  const started = await settle(() => {
    const e = host.call("session.operation", { thread: A, op: "clear-1" });
    return e.start === "started" && e;
  }, 30000);
  assert.equal(started.start, "started");
  assert.deepEqual(events(f, "launch").map((e) => e.session), [N, S]);
  const lines = host.logs.filter((l) => l.startsWith("clear clear-1:"));
  assert.deepEqual(lines, [`clear clear-1: successor ${S} (assistant): start held: capacity-full`]);
  assert.deepEqual(events(f, "refused").map((e) => [e.session, e.code]), [[S, "capacity-full"]]);
});

// sage-2830, Q2: after a restart, an earlier host's run of the cleared session that outlived
// recover() was ended host-lost, but it still runs. Its exit, not that end, is the proof, and it
// isn't signalled. A sleep child stands in for its runner, as in session-launch.test.mjs's Q2 test.
test("REQ-THREAD-2 T4 D7 Q2: after a restart, a live survivor of the cleared run holds the clear at drained with no signal, and its exit lets it retire", { skip, timeout: 90000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  const one = await start(t, prepared);
  const { A, D } = thread(one.call);
  const pid = 2 ** 22 + 7;
  const record = { business: "acme", role: "assistant", run: "run-left", harness: "pi", pid, startTime: "1", thread: A, session: D };
  const left = one.runtime.bindLaunch(record);
  one.runtime.broker.request(left.cap, { verb: "role.claim" });
  one.call("session.clear", { thread: A, session: D, revision: 0, confirm: true, op: "clear-1" });
  assert.deepEqual(one.runtime.broker.request(left.cap, { verb: "message.receive" }), []);
  assert.deepEqual(one.runtime.sessionClears("acme").map((c) => [c.stage, c.run]), [["drained", "run-left"]]);
  await one.close();

  const runner = spawn("sleep", ["60"], { stdio: "ignore" });
  const gone = new Promise((r) => runner.once("exit", r));
  t.after(() => runner.kill("SIGKILL"));
  const runnerStartTime = await until(() => startTimeOf(runner.pid));
  new Registry(f.dataRoot).add({ business: "acme", instance: "assistant", thread: A, session: D, run: "run-left", harness: "pi", pid, startTime: "1", runnerPid: runner.pid, runnerStartTime });
  const two = await start(t, prepared);
  assert.ok(two.logs.includes(`run run-left (assistant) from an earlier host ended: host-lost`), two.logs.join("\n"));
  // Ended in the bus, live on the host: the clear waits, and nothing is signalled.
  for (let i = 0; i < 5; i++) await two.launcher.tick();
  assert.deepEqual(two.runtime.sessionClears("acme").map((c) => c.stage), ["drained"]);
  assert.deepEqual(events(f, "session-clear-stop"), []);
  assert.equal(startTimeOf(runner.pid), runnerStartTime, "the survivor wasn't signalled");

  runner.kill("SIGKILL");
  await gone;
  const done = await settle(() => {
    const e = two.call("session.operation", { thread: A, op: "clear-1" });
    return e.start === "started" && e;
  }, 30000);
  assert.deepEqual([done.state, done.start], ["retired", "started"]);
  assert.deepEqual(events(f, "session-clear-stop"), []);
});

// D7 step 3: a drained run that doesn't exit within the stop timeout. The ticks are the test's, so
// its runner is stopped from outside its namespace before the clear's one TERM, as
// launcher.test.mjs's SIGTERM test does, and the TERM waits. The clear records uncertain once,
// sends nothing more, and retires when the run's end comes.
test("REQ-THREAD-2 T4 D7: a drained run that outlives the stop timeout makes the clear uncertain once, with one signal, and it retires on the run's end", { skip, timeout: 90000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  const host = await start(t, prepared, { tickMs: 3600000, stopTimeoutMs: 1000 });
  const tick = () => host.launcher.tick();
  const { A, D } = thread(host.call);
  const first = send(host.call, A, D, "before the clear");
  await tick();
  await first.reply();
  const [s] = readSessions(f.dataRoot);
  const kill = () => {
    if (startTimeOf(s.runnerPid) === s.runnerStartTime) process.kill(s.runnerPid, "SIGKILL");
  };
  t.after(kill);
  host.call("session.clear", { thread: A, session: D, revision: 0, confirm: true, op: "clear-1" });
  // The run's own receive drains it; no tick is needed for that.
  const drained = await settle(() => host.runtime.sessionClears("acme").find((c) => c.stage === "drained"));
  assert.equal(drained?.run, s.run);
  process.kill(s.runnerPid, "SIGSTOP");
  await tick();
  assert.equal(events(f, "session-clear-stop").length, 1);
  const uncertain = await settle(async () => {
    await tick();
    return host.runtime.sessionClears("acme").find((c) => c.stage === "uncertain");
  }, 15000);
  assert.equal(uncertain?.run, s.run);
  for (let i = 0; i < 5; i++) await tick();
  assert.equal(events(f, "session-clear-stop").length, 1);
  assert.equal(events(f, "session-clear-uncertain").length, 1);
  assert.equal(startTimeOf(s.runnerPid), s.runnerStartTime, "nothing forced it");
  process.kill(s.runnerPid, "SIGCONT");
  const done = await settle(async () => {
    await tick();
    const e = host.call("session.operation", { thread: A, op: "clear-1" });
    return e.start === "started" && e;
  }, 30000);
  assert.deepEqual([done.state, done.start], ["retired", "started"]);
  assert.equal(events(f, "session-clear-stop").length, 1);
  assert.equal(events(f, "session-clear-uncertain").length, 1);
});

// T6 (sage-3001): the bus that proves the stop is configured but unavailable. Its host's op
// rejects as host.mjs does when the broker exits before it replies (broker-exited), first at the
// exit proof, then at retirement. While it is down nothing retires, no successor launches and the
// cleared session's slot isn't handed on; each failure is logged once, and the clear resumes where
// it stopped once the bus answers again.
test("REQ-THREAD-2 T4 T6: a bus unavailable at the exit proof and then at retirement holds the clear with no retirement, successor or launch, and it resumes when the bus is back", { skip, timeout: 90000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  const host = await start(t, prepared);
  const down = new Set();
  const op = host.host.op;
  host.host.op = (m) => (down.has(m.op) ? Promise.reject(Object.assign(new Error("broker exited before it replied"), { code: "broker-exited" })) : op(m));
  const { A, D } = thread(host.call);
  await send(host.call, A, D, "before the clear").reply();
  const stage = () => host.runtime.sessionClears("acme").map((c) => c.stage);
  const lines = () => host.logs.filter((l) => l.startsWith("clear clear-1") || l.startsWith("thread tick:"));
  const line = `clear clear-1 (session ${D}): broker-exited`;
  const held = () => {
    assert.deepEqual(events(f, "session-clear-retired"), []);
    assert.deepEqual(events(f, "session-clear-failed"), []);
    assert.deepEqual(events(f, "launch").map((e) => e.session), [D]);
    assert.deepEqual(readSessions(f.dataRoot), [], "nothing runs in the cleared session's slot");
    assert.deepEqual(host.call("session.list", { thread: A }).map((r) => [r.id, r.default]), [[D, true]]);
  };

  down.add("sessionClearExited");
  const S = host.call("session.clear", { thread: A, session: D, revision: 0, confirm: true, op: "clear-1" }).successor;
  // The drained run is stopped and ends; the exit proof is what fails.
  await settle(() => events(f, "end").length === 1 && host.logs.includes(line));
  await new Promise((r) => setTimeout(r, 1000));
  assert.deepEqual(lines(), [line]);
  assert.deepEqual(stage(), ["drained"]);
  assert.equal(events(f, "session-clear-stop").length, 1);
  held();

  down.clear();
  down.add("sessionClearRetire");
  await settle(() => stage()[0] === "exited");
  await new Promise((r) => setTimeout(r, 1000));
  assert.deepEqual(stage(), ["exited"]);
  assert.deepEqual(lines(), [line], "the same code isn't logged again");
  held();

  down.clear();
  const done = await settle(() => {
    const e = host.call("session.operation", { thread: A, op: "clear-1" });
    return e.start === "started" && e;
  }, 30000);
  assert.deepEqual([done.state, done.start, done.refusal], ["retired", "started", null]);
  const kinds = launchLog(f).map((e) => [e.event, e.session ?? null]).filter(([e]) => ["session-clear-stop", "end", "session-clear-retired", "launch"].includes(e));
  assert.deepEqual(kinds, [
    ["launch", D],
    ["session-clear-stop", D],
    ["end", D],
    ["session-clear-retired", D],
    ["launch", S],
  ]);
  assert.deepEqual(lines(), [line]);
  assert.deepEqual(host.call("session.list", { thread: A }).map((r) => [r.id, r.state, r.default]), [
    [D, "retired", false],
    [S, "active", true],
  ]);
});