// Row 270 round 2 (264c, #1775): the read budget (B1, duckworth-0223, sage-3031,
// mos-1352) and Claude history from before the ledger (B3). Round 3: the cap on
// streams one collection opens (B4, duckworth-0226). The streams are Pi
// 0.85.1's shapes, written by hand as in compactions.test.mjs. A stream over
// the budget is a sparse file: truncate() makes the hole, so no test writes
// the bytes it reads past.

import { test } from "node:test";
import assert from "node:assert/strict";
import { appendFileSync, chmodSync, mkdtempSync, readFileSync, rmSync, truncateSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { collectCompactions, LEDGER, READ_BYTES, readBudget, readStreams, STREAMS } from "../src/compactions.mjs";
import { sessionStore } from "../src/session.mjs";
import { compactions as pi } from "../../plugins/harness/pi/index.mjs";

const MiB = 1 << 20;

function setup(t) {
  const root = mkdtempSync(join(tmpdir(), "seat-compactions-budget-"));
  t.after(() => rmSync(root, { recursive: true, force: true }));
  const store = sessionStore(root, "acme", "th1", "s1");
  let tick = 0;
  const now = () => new Date(Date.UTC(2026, 9, 22, 12, 0, tick++)).toISOString();
  const collect = (harness = "pi", budget = undefined, streams = undefined) =>
    collectCompactions(store, { name: harness, compactions: { pi }[harness] }, { now, ...(budget === undefined ? {} : { budget }), ...(streams === undefined ? {} : { streams }) });
  const records = () => readFileSync(join(store.dir, LEDGER), "utf8").trim().split("\n").map((l) => JSON.parse(l));
  return { root, store, collect, records };
}

const header = (id) => ({ type: "session", version: 3, id, timestamp: "2026-10-22T11:00:00.000Z", cwd: "/w" });
const compaction = (id, timestamp) => ({ type: "compaction", id, parentId: null, timestamp, summary: "s", firstKeptEntryId: "k", tokensBefore: 100 });
const line = (e) => `${JSON.stringify(e)}\n`;
const at = (n) => `2026-10-22T12:00:0${n}.000Z`;

test("the budget is the setting's whole number of bytes from 1 MiB to 1 GiB, and the default for anything else", () => {
  assert.equal(READ_BYTES, 64 * MiB);
  for (const v of [MiB, 8 * MiB, 1024 * MiB]) assert.equal(readBudget(v), v);
  for (const v of [undefined, null, "8388608", MiB + 0.5, -1, 0, MiB - 1, 1024 * MiB + 1, NaN, Infinity, {}, [MiB], true]) {
    assert.equal(readBudget(v), READ_BYTES, JSON.stringify(v) ?? String(v));
  }
});

test("a stream past the default budget is not-collected with a lower bound, and a later collection catches up", (t) => {
  const f = setup(t);
  const path = join(f.store.session, "a.jsonl");
  const whole = line(header("a")) + line(compaction("c1", "2026-10-22T11:01:00.000Z"));
  writeFileSync(path, whole);
  // A hole past the default budget, with no newline in it: a line still being written.
  truncateSync(path, READ_BYTES + 2 * MiB);
  assert.deepEqual(f.collect(), { count: 1, known: false, reason: "not-collected", at: at(0) });
  // The cursor stops at the last whole line read.
  assert.deepEqual(f.records().filter((r) => r.kind === "cursor").map((r) => r.bytes), [Buffer.byteLength(whole)]);
  // The long line ends up as c2: the bytes past the cursor are read again, from the cursor.
  truncateSync(path, Buffer.byteLength(whole));
  appendFileSync(path, line(compaction("c2", "2026-10-22T11:02:00.000Z")));
  assert.deepEqual(f.collect(), { count: 2, known: true, reason: null, at: at(1) });
});

test("a budget smaller than the streams counts over several collections, once each, and is known when it catches up", (t) => {
  const f = setup(t);
  const h = Buffer.byteLength(line(header("a"))), c = Buffer.byteLength(line(compaction("c1", "2026-10-22T11:01:00.000Z")));
  writeFileSync(join(f.store.session, "a.jsonl"), line(header("a")) + line(compaction("c1", "2026-10-22T11:01:00.000Z")) + line(compaction("c2", "2026-10-22T11:02:00.000Z")));
  writeFileSync(join(f.store.session, "b.jsonl"), line(header("b")) + line(compaction("c3", "2026-10-22T11:03:00.000Z")));
  // One byte past a's first compaction: a stops there and b isn't reached; then all of c2 and b's
  // header; then c3.
  const budget = h + c + 1;
  assert.deepEqual(f.collect("pi", budget), { count: 1, known: false, reason: "not-collected", at: at(0) });
  assert.deepEqual(f.collect("pi", budget), { count: 2, known: false, reason: "not-collected", at: at(1) });
  assert.deepEqual(f.collect("pi", budget), { count: 3, known: true, reason: null, at: at(2) });
  assert.deepEqual(f.records().filter((r) => r.kind === "compaction").map((r) => [r.entry, r.stream]), [["c1", "a.jsonl"], ["c2", "a.jsonl"], ["c3", "b.jsonl"]]);
});

test("a line split across reads counts once", (t) => {
  const f = setup(t);
  // A message that ends just short of the first read's end, so c1 crosses it.
  const first = line(header("a"));
  const pad = { type: "message", id: "m1", parentId: null, timestamp: "2026-10-22T11:00:01.000Z", message: { role: "user", content: "" } };
  pad.message.content = "x".repeat(MiB - 100 - Buffer.byteLength(first) - Buffer.byteLength(line(pad)));
  writeFileSync(join(f.store.session, "a.jsonl"), first + line(pad) + line(compaction("c1", "2026-10-22T11:01:00.000Z")));
  assert.equal(Buffer.byteLength(first + line(pad)), MiB - 100);
  assert.deepEqual(f.collect(), { count: 1, known: true, reason: null, at: at(0) });
});

// B3 (Vikunja 466): Claude Code runs before this ledger left history in claude-projects/ that
// can't be counted. The first collection is harness-unsupported, and it stays so.
test("Claude history from before the ledger makes the first Pi collection harness-unsupported, for good", (t) => {
  const f = setup(t);
  writeFileSync(join(f.store.projects, "old.jsonl"), "{}\n");
  writeFileSync(join(f.store.session, "a.jsonl"), line(header("a")) + line(compaction("c1", "2026-10-22T11:01:00.000Z")));
  assert.deepEqual(f.collect(), { count: null, known: false, reason: "harness-unsupported", at: at(0) });
  assert.deepEqual(f.collect(), { count: 1, known: false, reason: "harness-unsupported", at: at(1) });
});

test("an empty claude-projects/ is no Claude history, and one filled after the ledger exists isn't checked", (t) => {
  const f = setup(t);
  writeFileSync(join(f.store.session, "a.jsonl"), line(header("a")) + line(compaction("c1", "2026-10-22T11:01:00.000Z")));
  assert.deepEqual(f.collect(), { count: 1, known: true, reason: null, at: at(0) });
  // A Claude run after the ledger exists collects for itself (compactions.test.mjs).
  writeFileSync(join(f.store.projects, "later.jsonl"), "{}\n");
  assert.deepEqual(f.collect(), { count: 1, known: true, reason: null, at: at(1) });
});

test("a claude-projects/ that can't be read records nothing, and the next collection checks again", (t) => {
  const f = setup(t);
  writeFileSync(join(f.store.projects, "old.jsonl"), "{}\n");
  chmodSync(f.store.projects, 0o000);
  try {
    assert.throws(() => f.collect(), { code: "EACCES" });
    assert.throws(() => readFileSync(join(f.store.dir, LEDGER)), { code: "ENOENT" });
  } finally {
    chmodSync(f.store.projects, 0o700);
  }
  assert.deepEqual(f.collect(), { count: null, known: false, reason: "harness-unsupported", at: at(1) });
});

test("the stream cap is the setting's whole number from 32 to 65536, and the default for anything else", () => {
  assert.equal(STREAMS, 1024);
  for (const v of [32, 2048, 65536]) assert.equal(readStreams(v), v);
  for (const v of [undefined, null, "2048", 40.5, -1, 0, 31, 65537, NaN, Infinity, {}, [64], true]) {
    assert.equal(readStreams(v), STREAMS, JSON.stringify(v) ?? String(v));
  }
});

test("more streams than the cap is not-collected, with cursors only for the streams opened, and a larger cap catches up", (t) => {
  const f = setup(t);
  const names = ["a", "b", "c", "d"];
  names.forEach((n, i) => writeFileSync(join(f.store.session, `${n}.jsonl`), line(header(n)) + line(compaction(`c${n}`, `2026-10-22T11:0${i}:00.000Z`))));
  const cursors = () => f.records().filter((r) => r.kind === "cursor").map((r) => r.stream);
  assert.deepEqual(f.collect("pi", undefined, 3), { count: 3, known: false, reason: "not-collected", at: at(0) });
  assert.deepEqual(cursors(), ["a.jsonl", "b.jsonl", "c.jsonl"]);
  // The same cap again opens the same three, which haven't moved: no new lines but the collected one.
  assert.deepEqual(f.collect("pi", undefined, 3), { count: 3, known: false, reason: "not-collected", at: at(1) });
  assert.deepEqual(cursors(), ["a.jsonl", "b.jsonl", "c.jsonl"]);
  assert.deepEqual(f.collect("pi", undefined, 4), { count: 4, known: true, reason: null, at: at(2) });
  assert.deepEqual(cursors(), ["a.jsonl", "b.jsonl", "c.jsonl", "d.jsonl"]);
});

test("the default cap opens STREAMS streams and writes no more cursor lines than that", (t) => {
  const f = setup(t);
  for (let i = 0; i <= STREAMS; i++) writeFileSync(join(f.store.session, `e${String(i).padStart(4, "0")}.jsonl`), "");
  assert.deepEqual(f.collect(), { count: null, known: false, reason: "not-collected", at: at(0) });
  const cursors = f.records().filter((r) => r.kind === "cursor");
  assert.equal(cursors.length, STREAMS);
  assert.equal(cursors.at(-1).stream, "e1023.jsonl");
});