// Compaction counts on a real run (row 270, 264c, #1775, T3's Pi half): the
// launcher collects a thread session's completed compactions into its
// store's ledger at run start and at run end, and session.list reads the
// ledger through the runtime's reader. The count holds across a resume and a
// host restart. Real runners in the bwrap sandbox; a stub Pi adapter writes
// the session stream (fixtures/compaction-adapter.mjs).
//
// Round 2 (duckworth-0223 B2, sage-3031, mos-1352): a failed collection, the
// read budget from the system config, a clear's retired session and its
// successor, and a legacy session read through the runtime's reader. Round 3
// (duckworth-0226 B4, sage-3038): the stream cap from the system config.
//
// The host is session-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 { spawnSync } from "node:child_process";
import { existsSync, mkdirSync, readFileSync, rmSync, truncateSync, 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 { collectCompactions, LEDGER } from "../../seat/src/compactions.mjs";
import { gitEnv } from "../../seat/src/checkout.mjs";
import { sandboxProblem } from "../../seat/src/sandbox.mjs";
import { launchLogFile, sessionStore } from "../../seat/src/session.mjs";
import { compactions as pi } from "../../plugins/harness/pi/index.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/compaction-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 a wait a mutant could
// stall returns what it last saw and the assertions after it decide.
async function settle(fn, ms = 5000) {
  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, on Pi.
function prepare(t) {
  const root = tmp(t, "mosaic-compactions-");
  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: 3 });
    doc.launch.threadIdleSeconds = 1;
    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.
async function start(t, { f, adapter }) {
  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),
    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: 100,
    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 { 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 counted = (f) => events(f, "compactions").map((e) => [e.session, e.count, e.known, e.reason]);

// Send to a thread's default session and wait for the assistant's RESULT reply.
async function ask(call, thread, body) {
  const sent = call("thread.send", { thread, body });
  const r = await until(() => call("thread.messages", { thread }).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]);
}

const line = (e) => `${JSON.stringify(e)}\n`;

test("REQ-THREAD-2 T3 a run collects compactions at its start and its end, and the count holds across a resume and a host restart", { skip, timeout: 90000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  let host = await start(t, prepared);
  const project = host.call("project.create", { name: "stack work", repo: "stack" }).id;
  const A = host.call("thread.create", { project, name: "alpha" }).id;
  const [{ id: D }] = host.call("session.list", { thread: A });
  // An earlier run left a compaction no run-end collection saw (a host that died mid-run).
  // Only the run-start collection counts it before the agent's first turn.
  const store = sessionStore(f.dataRoot, "acme", A, D);
  writeFileSync(join(store.session, "pi.jsonl"), line({ type: "session", version: 3, id: "pi", timestamp: "2026-10-22T11:00:00.000Z", cwd: "/w" }) + line({ type: "compaction", id: "c0", parentId: null, timestamp: "2026-10-22T10:59:00.000Z", summary: "s", firstKeptEntryId: "k", tokensBefore: 100 }), { mode: 0o600 });

  // Run 1: start counts c0 (M13), the turn compacts c1, the end counts it (M14).
  assert.deepEqual(await ask(host.call, A, "compact c1"), { body: "compact c1", before: [] });
  await until(() => events(f, "end").length === 1, 30000);
  await settle(() => counted(f).length >= 2);
  assert.deepEqual(counted(f), [[D, 1, true, null], [D, 2, true, null]]);

  // A restart: a new host and launcher on the same data root. Run 2 resumes the session.
  await host.close();
  host = await start(t, prepared);
  assert.deepEqual(await ask(host.call, A, "compact c2"), { body: "compact c2", before: ["compact c1"] });
  await until(() => events(f, "end").length === 2, 30000);
  await settle(() => counted(f).length >= 4);
  assert.deepEqual(counted(f), [[D, 1, true, null], [D, 2, true, null], [D, 2, true, null], [D, 3, true, null]]);

  // session.list reads the ledger through the runtime's reader (M15), after another restart.
  await host.close();
  host = await start(t, prepared);
  const [row] = host.call("session.list", { thread: A });
  assert.equal(row.id, D);
  assert.deepEqual({ ...row.compactions, at: typeof row.compactions?.at }, { count: 3, known: true, reason: null, at: "string" });
});

test("REQ-THREAD-2 T3 a collection that fails is logged and recorded store-unreadable, and the run ends as one whose collections count", { skip, timeout: 90000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  const host = await start(t, prepared);
  const project = host.call("project.create", { name: "stack work", repo: "stack" }).id;
  const A = host.call("thread.create", { project, name: "alpha" }).id;
  const [{ id: D }] = host.call("session.list", { thread: A });
  // The ledger's path is a directory, so readLedger refuses it (ledger-unsafe).
  const store = sessionStore(f.dataRoot, "acme", A, D);
  mkdirSync(join(store.dir, LEDGER));
  assert.deepEqual(await ask(host.call, A, "compact c1"), { body: "compact c1", before: [] });
  await until(() => events(f, "end").length === 1, 30000);
  await settle(() => counted(f).length >= 2);
  assert.deepEqual(counted(f), [[D, null, false, "store-unreadable"], [D, null, false, "store-unreadable"]]);
  const [{ run: first }] = events(f, "launch");
  assert.equal(host.logs.filter((l) => l.startsWith(`run ${first}: compactions: `)).length, 2);
  // With the ledger's path free again, the next run's collections count what the stream holds.
  rmSync(join(store.dir, LEDGER), { recursive: true });
  assert.deepEqual(await ask(host.call, A, "compact c2"), { body: "compact c2", before: ["compact c1"] });
  await until(() => events(f, "end").length === 2, 30000);
  await settle(() => counted(f).length >= 4);
  assert.deepEqual(counted(f).slice(2), [[D, 1, true, null], [D, 2, true, null]]);
  const ends = events(f, "end").map((e) => [e.reason, e.exitCode, e.signal]);
  assert.deepEqual(ends[0], ends[1]);
});

test("REQ-THREAD-2 T3 the read budget is the system config's compactionReadBytes, and a bad value takes the default", { skip, timeout: 90000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  const setting = (v) => {
    const doc = JSON.parse(readFileSync(f.config, "utf8"));
    doc.compactionReadBytes = v;
    writeFileSync(f.config, JSON.stringify(doc));
  };
  setting(1 << 20);
  let host = await start(t, prepared);
  const project = host.call("project.create", { name: "stack work", repo: "stack" }).id;
  const A = host.call("thread.create", { project, name: "alpha" }).id;
  const [{ id: D }] = host.call("session.list", { thread: A });
  // c0, then a 2 MiB hole with no newline: a line past 1 MiB, inside the 64 MiB default.
  const store = sessionStore(f.dataRoot, "acme", A, D);
  const path = join(store.session, "pi.jsonl");
  const whole = line({ type: "session", version: 3, id: "pi", timestamp: "2026-10-22T11:00:00.000Z", cwd: "/w" }) + line({ type: "compaction", id: "c0", parentId: null, timestamp: "2026-10-22T10:59:00.000Z", summary: "s", firstKeptEntryId: "k", tokensBefore: 100 });
  writeFileSync(path, whole, { mode: 0o600 });
  truncateSync(path, Buffer.byteLength(whole) + 2 * (1 << 20));
  assert.deepEqual(await ask(host.call, A, "one"), { body: "one", before: [] });
  await until(() => events(f, "end").length === 1, 30000);
  await settle(() => counted(f).length >= 2);
  assert.deepEqual(counted(f), [[D, 1, false, "not-collected"], [D, 1, false, "not-collected"]]);
  assert.deepEqual(host.logs.filter((l) => l.startsWith("system config: ")), []);

  setting("x");
  await host.close();
  host = await start(t, prepared);
  assert.deepEqual(host.logs.filter((l) => l.startsWith("system config: ")), [
    'system config: compactionReadBytes "x" isn\'t a whole number of bytes from 1048576 to 1073741824; compaction collections read up to the default 67108864',
  ]);
  assert.deepEqual(await ask(host.call, A, "two"), { body: "two", before: ["one"] });
  await until(() => events(f, "end").length === 2, 30000);
  await settle(() => counted(f).length >= 4);
  assert.deepEqual(counted(f).slice(2), [[D, 1, true, null], [D, 1, true, null]]);
});

test("REQ-THREAD-2 T3 the stream cap is the system config's compactionStreams, and a bad value takes the default", { skip, timeout: 90000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  const setting = (v) => {
    const doc = JSON.parse(readFileSync(f.config, "utf8"));
    doc.compactionStreams = v;
    writeFileSync(f.config, JSON.stringify(doc));
  };
  setting(32);
  let host = await start(t, prepared);
  const project = host.call("project.create", { name: "stack work", repo: "stack" }).id;
  const A = host.call("thread.create", { project, name: "alpha" }).id;
  const [{ id: D }] = host.call("session.list", { thread: A });
  // 33 empty streams that sort before pi.jsonl: 34 in all, past a cap of 32, so pi.jsonl and its
  // c0 aren't opened. The default cap opens all 34.
  const store = sessionStore(f.dataRoot, "acme", A, D);
  for (let i = 0; i < 33; i++) writeFileSync(join(store.session, `e${String(i).padStart(2, "0")}.jsonl`), "", { mode: 0o600 });
  writeFileSync(join(store.session, "pi.jsonl"), line({ type: "session", version: 3, id: "pi", timestamp: "2026-10-22T11:00:00.000Z", cwd: "/w" }) + line({ type: "compaction", id: "c0", parentId: null, timestamp: "2026-10-22T10:59:00.000Z", summary: "s", firstKeptEntryId: "k", tokensBefore: 100 }), { mode: 0o600 });
  assert.deepEqual(await ask(host.call, A, "one"), { body: "one", before: [] });
  await until(() => events(f, "end").length === 1, 30000);
  await settle(() => counted(f).length >= 2);
  assert.deepEqual(counted(f), [[D, null, false, "not-collected"], [D, null, false, "not-collected"]]);
  assert.deepEqual(host.logs.filter((l) => l.startsWith("system config: ")), []);

  setting("x");
  await host.close();
  host = await start(t, prepared);
  assert.deepEqual(host.logs.filter((l) => l.startsWith("system config: ")), [
    'system config: compactionStreams "x" isn\'t a whole number from 32 to 65536; compaction collections open up to the default 1024 streams',
  ]);
  assert.deepEqual(await ask(host.call, A, "two"), { body: "two", before: ["one"] });
  await until(() => events(f, "end").length === 2, 30000);
  await settle(() => counted(f).length >= 4);
  assert.deepEqual(counted(f).slice(2), [[D, 1, true, null], [D, 1, true, null]]);
});

test("REQ-THREAD-2 T3 T4 a clear leaves the retired session's count as it was, its successor counts from 0, and a legacy session reads the thread's ledger", { skip, timeout: 90000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  let host = await start(t, prepared);
  const project = host.call("project.create", { name: "stack work", repo: "stack" }).id;
  const A = host.call("thread.create", { project, name: "alpha" }).id;
  const [{ id: D }] = host.call("session.list", { thread: A });
  assert.deepEqual(await ask(host.call, A, "compact c1"), { body: "compact c1", before: [] });
  await until(() => events(f, "end").length === 1, 30000);
  await settle(() => counted(f).length >= 2);
  assert.deepEqual(counted(f), [[D, 0, true, null], [D, 1, true, null]]);

  // The run has ended, so the clear retires D at once and starts the successor, whose run-start
  // collection is its first.
  const S = host.call("session.clear", { thread: A, session: D, revision: 0, confirm: true, op: "clear-1" }).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], ["retired", "started"]);
  await settle(() => counted(f).some(([s]) => s === S));
  assert.deepEqual(counted(f).filter(([s]) => s === S)[0], [S, 0, true, null]);
  const list = () => Object.fromEntries(host.call("session.list", { thread: A }).map((r) => [r.id, r.compactions && { count: r.compactions.count, known: r.compactions.known, reason: r.compactions.reason }]));
  assert.deepEqual(list(), { [D]: { count: 1, known: true, reason: null }, [S]: { count: 0, known: true, reason: null } });

  // A legacy session (op legacy-3l, from the 3k-3l migration): its store is the thread's, made by
  // hand as packages/bus/tests/session-compactions.test.mjs does, with a ledger the launcher's
  // collect wrote.
  await until(() => events(f, "end").length === 2, 30000);
  await host.close();
  const L = "4b1c3a52-9a51-4d7e-8f2e-6c0a1b2d3e4f";
  const seeded = new Store(f.dataRoot);
  const at = "2026-10-21T00:00:00.000Z";
  seeded.run("INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)", L, at, "acme", project, "assistant", f.doc.human, "legacy");
  seeded.run("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=?", L);
  seeded.run("INSERT INTO session_defaults(at,thread,session,by) SELECT at,id,id,'migration-3l' FROM threads WHERE id=?", L);
  seeded.close();
  const legacy = sessionStore(f.dataRoot, "acme", L, L, { legacy: true });
  writeFileSync(join(legacy.session, "pi.jsonl"), line({ type: "session", version: 3, id: "pi", timestamp: "2026-10-22T11:00:00.000Z", cwd: "/w" }) + line({ type: "compaction", id: "c7", parentId: null, timestamp: "2026-10-22T11:07:00.000Z", summary: "s", firstKeptEntryId: "k", tokensBefore: 100 }), { mode: 0o600 });
  assert.equal(collectCompactions(legacy, { name: "pi", compactions: pi }).count, 1);
  host = await start(t, prepared);
  const [row] = host.call("session.list", { thread: L });
  assert.equal(row.id, L);
  assert.deepEqual({ ...row.compactions, at: typeof row.compactions?.at }, { count: 1, known: true, reason: null, at: "string" });
  assert.deepEqual(list(), { [D]: { count: 1, known: true, reason: null }, [S]: { count: 0, known: true, reason: null } });
});