// Session runs (row 269, 264b, #1774): every session of a thread runs on its
// own, with its own store under threads/<b>/<t>/sessions/<s>/ and its own
// workspace under session-workspaces/<b>/<instance>/<s>/, and a lock beside
// the workspace keeps it to one writer. A legacy session (op legacy-3l) keeps
// the thread's old store, and its workspace shows inside the sandbox at the
// old instance path. Real runners in the bwrap sandbox; a stub adapter keeps
// a history in the store and answers with what it found, its cwd and what
// the cwd holds.
//
// The host is thread-launch.test.mjs's: startBroker on bootConfig's config,
// with the ops the broker child forwards, and the human capability from
// bindHuman directly.

import { test } from "node:test";
import assert from "node:assert/strict";
import { spawn, spawnSync } from "node:child_process";
import { chmodSync, existsSync, mkdirSync, readdirSync, readFileSync, symlinkSync, 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 { Store } from "../../bus/src/store.mjs";
import { transition } from "../../processing-control/src/state.mjs";
import { gitEnv } from "../../seat/src/checkout.mjs";
import { startTimeOf } from "../../seat/src/proc.mjs";
import { 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";
// A thread made by hand gets the default session the 3k-3l migration gives a legacy thread, with
// the thread's id (as in thread-launch.test.mjs).
const DEFAULT_SESSION = [
  "INSERT INTO sessions(at,business,thread,id,instance,owner,name,op) SELECT at,business,id,id,instance,owner,'default','legacy-3l' FROM threads WHERE id=?",
  "INSERT INTO session_defaults(at,thread,session,by) SELECT at,id,id,'migration-3l' FROM threads WHERE id=?",
];

const ADAPTER = fileURLToPath(new URL("./fixtures/session-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), so the waits a mutant could
// stall 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);
}

// Each directory made 0700, as the store and workspace checks want.
function privateDir(...parts) {
  for (let i = 1; i <= parts.length; i++) {
    const d = join(...parts.slice(0, i));
    if (!existsSync(d)) mkdirSync(d, { mode: 0o700 });
    if (i > 1) chmodSync(d, 0o700);
  }
  return join(...parts);
}

// The assistant is the thread instance; sonnet is its model's limit.
function prepare(t, { sonnet = 3, idle = 1 } = {}) {
  const root = tmp(t, "mosaic-session-");
  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 };
}

// A host and its launcher on the prepared root. `ops` replaces launch ops, given the runtime.
async function start(t, { f, adapter }, { tickMs = 100, ops: replace } = {}) {
  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 = {
    'request.admission': m => runtime.admission(m),
    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),
    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),
    ...replace?.(runtime),
  };
  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,
    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);
// The refusals once there are n, or once anything launched.
const outcome = async (f, n = 1) => {
  await settle(() => events(f, "refused").length >= n || events(f, "launch").length);
  return events(f, "refused").map((e) => [e.session, e.code]);
};
const historyOf = (p) => (existsSync(p) ? readFileSync(p, "utf8") : null);

// Send to one session of a thread; reply() waits for the assistant's RESULT reply in that session.
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 };
}

// A thread made with thread.create, its default session and a second named one.
function twoSessions(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 });
  const S = call("session.create", { thread: A, name: "second", op: "op-second" }).id;
  return { project, A, D, S };
}

test("REQ-THREAD-2 REQ-LAUNCH-1 two sessions of one thread, and another thread's default, run at once, each with its own store and workspace; a restart of one resumes only that one", { skip, timeout: 90000 }, async (t) => {
  const prepared = prepare(t, { idle: 5 });
  const { f } = prepared;
  const host = await start(t, prepared);
  const { project, A, D, S } = twoSessions(host.call);
  const B = host.call("thread.create", { project, name: "beta" }).id;
  const [{ id: BD }] = host.call("session.list", { thread: B });

  const toD = send(host.call, A, D, "first for D");
  const toS = send(host.call, A, S, "first for S");
  const toB = send(host.call, B, null, "first for B");
  // All three run together (M1, M2: a shared workspace would refuse the second start busy).
  await settle(() => readSessions(f.dataRoot).length === 3 || events(f, "refused").length);
  assert.deepEqual(events(f, "refused"), []);
  assert.deepEqual(readSessions(f.dataRoot).map((s) => [s.thread, s.session]).sort(), [[A, D], [A, S], [B, BD]].sort());
  const [sawD, sawS, sawB] = await Promise.all([toD.reply(), toS.reply(), toB.reply()]);
  // Each session's history is its own (M5), and each runs in its own workspace at its own path (M8).
  assert.deepEqual([sawD.body, sawD.before, sawD.cwd], ["first for D", [], sessionWs(f, D)]);
  assert.deepEqual([sawS.body, sawS.before, sawS.cwd], ["first for S", [], sessionWs(f, S)]);
  assert.deepEqual([sawB.body, sawB.before, sawB.cwd], ["first for B", [], sessionWs(f, BD)]);
  const store = (thread, session) => join(f.dataRoot, "threads", "acme", thread, "sessions", session, "session", "history");
  assert.deepEqual([historyOf(store(A, D)), historyOf(store(A, S)), historyOf(store(B, BD))], ["first for D\n", "first for S\n", "first for B\n"]);
  for (const thread of [A, B]) assert.equal(existsSync(join(f.dataRoot, "threads", "acme", thread, "session")), false);
  assert.equal(existsSync(join(f.dataRoot, "workspaces", "acme", "assistant")), false, "no thread session uses the old instance workspace");
  for (const e of events(f, "launch")) assert.ok([D, S, BD].includes(e.session), JSON.stringify(e));

  // All three stop when idle, and each lock goes with its run.
  await until(() => events(f, "end").length === 3, 30000);
  assert.deepEqual(readSessions(f.dataRoot), []);
  for (const s of [D, S, BD]) assert.equal(existsSync(`${sessionWs(f, s)}.lock`), false, s);

  // The named session's next message starts only that session (M9), which finds only its history.
  const again = await send(host.call, A, S, "second for S").reply();
  assert.deepEqual([again.before, again.cwd], [["first for S"], sessionWs(f, S)]);
  assert.deepEqual(events(f, "launch").slice(3).map((e) => [e.thread, e.session]), [[A, S]]);
  assert.equal(historyOf(store(A, D)), "first for D\n");
});

// D9 (M7): a thread from before 3l. Its legacy session resumes from the thread's old store, and
// its new workspace shows at the old instance path, which the transcripts recorded as the cwd.
// The old host workspace isn't mounted.
test("REQ-THREAD-2 REQ-LAUNCH-1 a legacy session resumes the thread's old store at the old in-sandbox cwd, in a workspace of its own", { skip, timeout: 60000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  const store = new Store(f.dataRoot);
  const at = () => new Date().toISOString();
  store.run("INSERT INTO projects(id,at,business,name,repo,created_by) VALUES(?,?,?,?,?,?)", "p-hand", at(), "acme", "hand made", "stack", "jason");
  store.run("INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)", "th-legacy", at(), "acme", "p-hand", "assistant", "jason", "old");
  for (const sql of DEFAULT_SESSION) store.run(sql, "th-legacy");
  store.close();
  const old = privateDir(f.dataRoot, "threads", "acme", "th-legacy");
  writeFileSync(join(privateDir(old, "session"), "history"), "old sentinel\n");
  privateDir(old, "claude-projects");
  writeFileSync(join(privateDir(f.dataRoot, "workspaces", "acme", "assistant"), "old-file"), "the 264a workspace\n");

  const host = await start(t, prepared);
  const saw = await send(host.call, "th-legacy", null, "after 3l").reply();
  assert.deepEqual(saw.before, ["old sentinel"]);
  assert.equal(saw.cwd, join(f.dataRoot, "workspaces", "acme", "assistant"));
  assert.equal(saw.entries.includes("old-file"), false, "the old host workspace isn't mounted");
  assert.equal(readFileSync(join(old, "session", "history"), "utf8"), "old sentinel\nafter 3l\n");
  assert.equal(existsSync(join(old, "sessions")), false);
  assert.ok(existsSync(sessionWs(f, "th-legacy")));
  assert.deepEqual(readdirSync(join(f.dataRoot, "workspaces", "acme", "assistant")), ["old-file"]);
  assert.deepEqual(events(f, "launch").map((e) => [e.thread, e.session]), [["th-legacy", "th-legacy"]]);
});

// T2, M3: another host's live lock on the session's workspace refuses the start before any
// checkout. Once that host is gone the lock is stale, and the next start takes it over.
test("REQ-THREAD-2 REQ-LAUNCH-1 a live lock on a session workspace refuses session-workspace-busy before any checkout; a stale one is taken over", { skip, timeout: 60000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  const host = await start(t, prepared);
  const { A, S } = twoSessions(host.call);
  const other = spawn("sleep", ["60"], { stdio: "ignore" });
  const gone = new Promise((r) => other.once("exit", r));
  t.after(() => other.kill("SIGKILL"));
  const otherStart = await until(() => startTimeOf(other.pid));
  privateDir(f.dataRoot, "session-workspaces", "acme", "assistant");
  const lock = `${sessionWs(f, S)}.lock`;
  const held = `${JSON.stringify({ hostPid: other.pid, hostStartTime: otherStart, business: "acme", instance: "assistant", thread: A, session: S })}\n`;
  writeFileSync(lock, held, { mode: 0o600 });

  const toS = send(host.call, A, S, "for S");
  assert.deepEqual(await outcome(f), [[S, "session-workspace-busy"]]);
  assert.equal(events(f, "refused")[0].thread, A);
  assert.deepEqual(events(f, "launch"), []);
  assert.deepEqual(readSessions(f.dataRoot), []);
  assert.deepEqual(readdirSync(sessionWs(f, S)), [], "no checkout");
  assert.equal(readFileSync(lock, "utf8"), held);

  other.kill("SIGKILL");
  await gone;
  const saw = await toS.reply();
  assert.deepEqual([saw.before, saw.cwd], [[], sessionWs(f, S)]);
  await until(() => events(f, "end").length === 1);
  assert.equal(existsSync(lock), false);
});

// Q2 (BUILD.md:177-179, t1-sessions.md:109): an earlier host's sessionless run of the instance is
// still live, so no session of it starts until that run has exited. A sleep child stands in for
// its runner; its outer pid is gone, so recover() has nothing to kill and keeps it as a survivor.
test("REQ-THREAD-2 REQ-LAUNCH-1 a session start refuses legacy-run-live while an earlier host's sessionless run lives, and starts once it has exited", { skip, timeout: 60000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  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: "th-old", run: "run-left", harness: "pi", pid: 2 ** 22 + 7, startTime: "gone", runnerPid: runner.pid, runnerStartTime });
  const host = await start(t, prepared);
  assert.ok(host.logs.includes(`run run-left (assistant) from an earlier host still runs (pid ${2 ** 22 + 7}, runner ${runner.pid}): no session of assistant starts until it exits`), host.logs.join("\n"));
  const { A, S } = twoSessions(host.call);

  const toS = send(host.call, A, S, "for S");
  assert.deepEqual(await outcome(f), [[S, "legacy-run-live"]]);
  assert.equal(events(f, "refused")[0].thread, A);
  assert.deepEqual(events(f, "launch"), [], "nothing spawned");
  assert.deepEqual(readSessions(f.dataRoot), []);
  assert.equal(existsSync(sessionWs(f, S)), false, "refused before the workspace");

  runner.kill("SIGKILL");
  await gone;
  assert.equal(await settle(() => events(f, "launch").length), 1, "it starts once the run has exited");
  const saw = await toS.reply();
  assert.deepEqual([saw.before, saw.cwd], [[], sessionWs(f, S)]);
  await until(() => events(f, "end").length === 1);
});

// T2, C3: a symlinked session root refuses session-workspace-unsafe, at the segment check.
test("REQ-THREAD-2 REQ-LAUNCH-1 a symlinked session workspace root refuses session-workspace-unsafe and launches nothing", { skip, timeout: 60000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  const elsewhere = privateDir(prepared.root, "elsewhere");
  symlinkSync(elsewhere, join(f.dataRoot, "session-workspaces"));
  const host = await start(t, prepared);
  const { A, D } = twoSessions(host.call);
  send(host.call, A, D, "for D");
  assert.deepEqual(await outcome(f), [[D, "session-workspace-unsafe"]]);
  assert.deepEqual(events(f, "launch"), []);
  assert.deepEqual(readdirSync(elsewhere), []);
});

// T2, M4: every segment of the session path is a real directory, but workspaces/<b> is a symlink
// to it, so only the realpath compare finds that the session workspace holds the legacy root.
test("REQ-THREAD-2 REQ-LAUNCH-1 a session workspace that holds the legacy workspace root refuses session-workspace-overlap and launches nothing", { skip, timeout: 60000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  const host = await start(t, prepared);
  const { A, D } = twoSessions(host.call);
  const ws = privateDir(f.dataRoot, "session-workspaces", "acme", "assistant", D);
  symlinkSync(ws, join(privateDir(f.dataRoot, "workspaces"), "acme"));
  send(host.call, A, D, "for D");
  assert.deepEqual(await outcome(f), [[D, "session-workspace-overlap"]]);
  assert.deepEqual(events(f, "launch"), []);
  assert.deepEqual(readdirSync(ws), []);
});

// T2: a symlink among the store's segments refuses thread-store-refused before the workspace.
test("REQ-THREAD-2 REQ-LAUNCH-1 a symlinked session store refuses thread-store-refused and makes no workspace", { skip, timeout: 60000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  const host = await start(t, prepared);
  const { A, D } = twoSessions(host.call);
  const elsewhere = privateDir(prepared.root, "elsewhere");
  symlinkSync(elsewhere, join(privateDir(f.dataRoot, "threads", "acme", A), "sessions"));
  send(host.call, A, D, "for D");
  assert.deepEqual(await outcome(f), [[D, "thread-store-refused"]]);
  assert.match(events(f, "refused")[0].why, /is unsafe$/);
  assert.deepEqual(events(f, "launch"), []);
  assert.deepEqual(readdirSync(elsewhere), []);
  assert.equal(existsSync(join(f.dataRoot, "session-workspaces")), false);
});

// T2, M6 and the halt fence (C2). The credentialStatus op runs in start() after check() returned.
// Its first call halts and resumes, a new generation with no halt, so only start()'s own
// admission(admissionGeneration) refuses, before the session workspace is made. On the retry,
// the recheck's mayLaunch op (inside check(), after the lock is taken) halts once it has answered,
// and the recheck's admission() refuses processing-halted. Neither attempt spawns, and the second releases its lock.
test("REQ-THREAD-2 REQ-LAUNCH-1 a named session's start refuses processing-generation-changed, then processing-halted, after check() and before any spawn", { skip, timeout: 60000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  let credentials = 0;
  let halts = 0;
  const halt = () => transition(f.dataRoot, { business: "acme", human: f.doc.human, halted: true });
  const host = await start(t, prepared, {
    ops: (runtime) => ({
      credentialStatus: (m) => {
        if (m.role === "assistant" && ++credentials === 1) {
          transition(f.dataRoot, { business: "acme", human: f.doc.human, halted: false, generation: halt().generation });
          halts++;
        }
        return runtime.credentialStatus(m.business, m.role);
      },
      // The halt lands after the op answers: halted inside it, the bus's own fence
      // (runtime.mjs processingWork) throws and the launcher refuses limit-check-failed.
      mayLaunch: async (m) => {
        const may = await runtime.mayLaunch(m);
        if (m.instance === "assistant" && credentials === 2 && halts === 1) {
          halt();
          halts++;
        }
        return may;
      },
    }),
  });
  const { A, S } = twoSessions(host.call);
  send(host.call, A, S, "for S");
  const refused = await outcome(f, 2);
  assert.deepEqual(refused, [[S, "processing-generation-changed"], [S, "processing-halted"]]);
  assert.deepEqual([credentials, halts], [2, 2], "check() passed both times");
  assert.equal(events(f, "launch").length, 0);
  assert.deepEqual(readSessions(f.dataRoot), []);
  assert.equal(existsSync(join(f.dataRoot, "launches", "acme")), false, "no run directory");
  assert.ok(existsSync(sessionWs(f, S)), "the retry made the workspace");
  assert.equal(existsSync(`${sessionWs(f, S)}.lock`), false, "and released its lock");
});