// Row 270 (264c, #1775, T1 item 7 D6, T3's Pi half): the compaction count
// from a session's native Pi store, through the host ledger
// (src/compactions.mjs). The streams are Pi 0.85.1's shapes, written by hand:
// a session header, then entries, one JSON object per line.

import { test } from "node:test";
import assert from "node:assert/strict";
import { appendFileSync, lstatSync, mkdirSync, mkdtempSync, readFileSync, rmSync, statSync, symlinkSync, unlinkSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join, relative, isAbsolute } from "node:path";
import { collectCompactions, compactionCount, LEDGER, readLedger, storeDir } from "../src/compactions.mjs";
import { binds } from "../src/sandbox.mjs";
import { sessionStore } from "../src/session.mjs";
import { compactions as pi } from "../../plugins/harness/pi/index.mjs";
import { compactions as claude } from "../../plugins/harness/claude-code/index.mjs";

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

const header = (id) => ({ type: "session", version: 3, id, timestamp: "2026-10-22T11:00:00.000Z", cwd: "/w" });
const message = (id, role = "user") => ({ type: "message", id, parentId: null, timestamp: "2026-10-22T11:00:01.000Z", message: { role, content: "x" } });
const compaction = (id, timestamp) => ({ type: "compaction", id, parentId: null, timestamp, summary: "s", firstKeptEntryId: "k", tokensBefore: 100 });
const line = (e) => `${JSON.stringify(e)}\n`;
const stream = (store, name, entries) => writeFileSync(join(store.session, name), entries.map(line).join(""));
const append = (store, name, ...entries) => appendFileSync(join(store.session, name), entries.map(line).join(""));

test("an empty store, a store with no compaction and a failed start count 0, known", (t) => {
  const f = setup(t);
  assert.deepEqual(f.collect(), { count: 0, known: true, reason: null, at: "2026-10-22T12:00:00.000Z" });
  // A started and aborted compaction writes no entry; a compactionSummary message, a
  // branch_summary and a message are not compactions.
  stream(f.store, "a.jsonl", [header("a"), message("m1"), message("m2", "compactionSummary"), { type: "branch_summary", id: "b1", parentId: null, timestamp: "2026-10-22T11:00:02.000Z", summary: "b" }]);
  assert.deepEqual(f.collect(), { count: 0, known: true, reason: null, at: "2026-10-22T12:00:01.000Z" });
});

test("one and three completed compactions", (t) => {
  const f = setup(t);
  stream(f.store, "a.jsonl", [header("a"), message("m1"), compaction("c1", "2026-10-22T11:01:00.000Z")]);
  assert.equal(f.collect().count, 1);
  append(f.store, "a.jsonl", message("m2"), compaction("c2", "2026-10-22T11:02:00.000Z"), compaction("c3", "2026-10-22T11:03:00.000Z"));
  assert.deepEqual(f.collect(), { count: 3, known: true, reason: null, at: "2026-10-22T12:00:01.000Z" });
  assert.equal(compactionCount(f.store.dir).count, 3);
});

// sage-2814: Pi's createBranchedSession and forkFrom copy entries into a new stream with
// their id and timestamp, so a copied compaction counts once.
test("a compaction copied into a branched stream counts once, under the stream it was first seen in", (t) => {
  const f = setup(t);
  const c1 = compaction("c1", "2026-10-22T11:01:00.000Z");
  stream(f.store, "a.jsonl", [header("a"), message("m1"), c1]);
  assert.equal(f.collect().count, 1);
  stream(f.store, "b.jsonl", [header("b"), message("m1"), c1, message("m3")]);
  assert.deepEqual(f.collect(), { count: 1, known: true, reason: null, at: "2026-10-22T12:00:01.000Z" });
  append(f.store, "b.jsonl", compaction("c2", "2026-10-22T11:05:00.000Z"));
  assert.equal(f.collect().count, 2);
  const lines = f.records().filter((r) => r.kind === "compaction");
  assert.deepEqual(lines.map((r) => [r.entry, r.stream]), [["c1", "a.jsonl"], ["c2", "b.jsonl"]]);
});

test("one collection over two streams holding the same copy counts it once and writes it once", (t) => {
  const f = setup(t);
  const c1 = compaction("c1", "2026-10-22T11:01:00.000Z");
  stream(f.store, "a.jsonl", [header("a"), c1]);
  stream(f.store, "b.jsonl", [header("b"), c1]);
  assert.equal(f.collect().count, 1);
  assert.equal(f.records().filter((r) => r.kind === "compaction").length, 1);
});

// Pi's ids are unique only within a file: the same id at another time is another compaction.
test("the same entry id at another time in another stream is another compaction", (t) => {
  const f = setup(t);
  stream(f.store, "a.jsonl", [header("a"), compaction("c1", "2026-10-22T11:01:00.000Z")]);
  stream(f.store, "b.jsonl", [header("b"), compaction("c1", "2026-10-22T11:09:00.000Z")]);
  assert.equal(f.collect().count, 2);
});

test("resume: each collection reads past its cursor and writes a key once", (t) => {
  const f = setup(t);
  stream(f.store, "a.jsonl", [header("a"), compaction("c1", "2026-10-22T11:01:00.000Z")]);
  f.collect();
  f.collect();
  append(f.store, "a.jsonl", compaction("c2", "2026-10-22T11:02:00.000Z"));
  assert.equal(f.collect().count, 2);
  const r = f.records();
  assert.deepEqual(r.filter((x) => x.kind === "compaction").map((x) => x.entry), ["c1", "c2"]);
  assert.deepEqual(r.filter((x) => x.kind === "cursor").map((x) => x.bytes), [statSync(join(f.store.session, "a.jsonl")).size - line(compaction("c2", "2026-10-22T11:02:00.000Z")).length, statSync(join(f.store.session, "a.jsonl")).size]);
  assert.equal(r.filter((x) => x.kind === "collected").length, 3);
  assert.equal(lstatSync(join(f.store.dir, LEDGER)).mode & 0o777, 0o600);
});

test("a partial trailing line waits for its newline", (t) => {
  const f = setup(t);
  stream(f.store, "a.jsonl", [header("a")]);
  const whole = line(compaction("c1", "2026-10-22T11:01:00.000Z"));
  appendFileSync(join(f.store.session, "a.jsonl"), whole.slice(0, 30));
  assert.deepEqual(f.collect(), { count: 0, known: true, reason: null, at: "2026-10-22T12:00:00.000Z" });
  appendFileSync(join(f.store.session, "a.jsonl"), whole.slice(30));
  assert.deepEqual(f.collect(), { count: 1, known: true, reason: null, at: "2026-10-22T12:00:01.000Z" });
});

test("a missing ledger is not-collected and an unreadable one store-unreadable, never 0", (t) => {
  const f = setup(t);
  assert.deepEqual(compactionCount(f.store.dir), { count: null, known: false, reason: "not-collected", at: null });
  mkdirSync(join(f.store.dir, LEDGER));
  assert.deepEqual(compactionCount(f.store.dir), { count: null, known: false, reason: "store-unreadable", at: null });
  assert.throws(() => f.collect());
});

test("a symlinked ledger is store-unreadable and is not written through", (t) => {
  const f = setup(t);
  const target = join(f.root, "elsewhere.jsonl");
  writeFileSync(target, "");
  symlinkSync(target, join(f.store.dir, LEDGER));
  assert.equal(compactionCount(f.store.dir).reason, "store-unreadable");
  assert.throws(() => f.collect());
  assert.equal(readFileSync(target, "utf8"), "");
});

// sage-2806 condition 3: a torn last line is a bad line, not a skipped one.
test("a torn last ledger line makes the count unknown, a lower bound, and stays so", (t) => {
  const f = setup(t);
  assert.equal(f.collect().count, 0);
  appendFileSync(join(f.store.dir, LEDGER), '{"v":1,"kind":"compac');
  assert.deepEqual(compactionCount(f.store.dir), { count: null, known: false, reason: "store-incomplete", at: "2026-10-22T12:00:00.000Z" });
  stream(f.store, "a.jsonl", [header("a"), compaction("c1", "2026-10-22T11:01:00.000Z")]);
  assert.deepEqual(f.collect(), { count: 1, known: false, reason: "store-incomplete", at: "2026-10-22T12:00:01.000Z" });
  // The next write started on a fresh line: every line after the torn one is whole.
  const lines = f.ledger().split("\n");
  assert.equal(lines.pop(), "");
  const bad = lines.filter((l) => {
    try {
      JSON.parse(l);
      return false;
    } catch {
      return true;
    }
  });
  assert.deepEqual(bad, ['{"v":1,"kind":"compac']);
  assert.equal(readLedger(f.store.dir).events.size, 1);
});

test("a bad ledger line in the middle makes the count unknown", (t) => {
  const f = setup(t);
  stream(f.store, "a.jsonl", [header("a"), compaction("c1", "2026-10-22T11:01:00.000Z")]);
  f.collect();
  appendFileSync(join(f.store.dir, LEDGER), "not json\n");
  append(f.store, "a.jsonl", compaction("c2", "2026-10-22T11:02:00.000Z"));
  assert.deepEqual(f.collect(), { count: 2, known: false, reason: "store-incomplete", at: "2026-10-22T12:00:01.000Z" });
});

test("a stream that shrank, changed under its cursor or went away is store-incomplete for good", async (t) => {
  for (const change of ["shrink", "rewrite", "remove"]) {
    await t.test(change, (t) => {
      const f = setup(t);
      stream(f.store, "a.jsonl", [header("a"), compaction("c1", "2026-10-22T11:01:00.000Z")]);
      assert.equal(f.collect().known, true);
      const path = join(f.store.session, "a.jsonl");
      if (change === "shrink") stream(f.store, "a.jsonl", [header("a")]);
      // Same length, other bytes: only the cursor's tail digest sees it.
      if (change === "rewrite") stream(f.store, "a.jsonl", [header("a"), compaction("c9", "2026-10-22T11:01:00.000Z")]);
      if (change === "remove") unlinkSync(path);
      const c = f.collect();
      assert.equal(c.known, false);
      assert.equal(c.reason, "store-incomplete");
      assert.equal(c.count, change === "rewrite" ? 2 : 1);
      // A later clean collection doesn't make it known again.
      assert.equal(f.collect().reason, "store-incomplete");
    });
  }
});

test("a stream that isn't a regular file or a Pi session is store-unreadable, and that passes", (t) => {
  const f = setup(t);
  const target = join(f.root, "outside.jsonl");
  writeFileSync(target, line(header("x")) + line(compaction("c1", "2026-10-22T11:01:00.000Z")));
  symlinkSync(target, join(f.store.session, "link.jsonl"));
  assert.deepEqual(f.collect(), { count: null, known: false, reason: "store-unreadable", at: "2026-10-22T12:00:00.000Z" });
  unlinkSync(join(f.store.session, "link.jsonl"));
  assert.deepEqual(f.collect(), { count: 0, known: true, reason: null, at: "2026-10-22T12:00:01.000Z" });
});

test("a first line that isn't a session header is store-unreadable", (t) => {
  const f = setup(t);
  stream(f.store, "a.jsonl", [compaction("c1", "2026-10-22T11:01:00.000Z")]);
  assert.equal(f.collect().reason, "store-unreadable");
});

// RULINGS Q6: Claude Code is unknown with the Vikunja 466 reason, never 0, and a session any
// Claude run collected for stays unknown.
test("a Claude run is harness-unsupported, and stays so after a Pi run", (t) => {
  const f = setup(t);
  assert.deepEqual(f.collect("claude"), { count: null, known: false, reason: "harness-unsupported", at: "2026-10-22T12:00:00.000Z" });
  stream(f.store, "a.jsonl", [header("a"), compaction("c1", "2026-10-22T11:01:00.000Z")]);
  assert.deepEqual(f.collect("pi"), { count: 1, known: false, reason: "harness-unsupported", at: "2026-10-22T12:00:01.000Z" });
});

test("storeDir is the store sessionStore makes, without making it", (t) => {
  const f = setup(t);
  assert.equal(storeDir(f.root, "acme", "th1", "s1"), f.store.dir);
  assert.equal(storeDir(f.root, "acme", "th1", "th1", { legacy: true }), join(f.root, "threads", "acme", "th1"));
  assert.equal(storeDir(f.root, "acme", "th1", "s2"), join(f.store.dir, "..", "s2"));
  assert.throws(() => lstatSync(join(f.store.dir, "..", "s2")), { code: "ENOENT" });
  for (const bad of ["..", "a/b", "", "A", null]) assert.equal(storeDir(f.root, "acme", "th1", bad), null);
  assert.equal(storeDir(f.root, "../x", "th1", "s1"), null);
});

// sage-2806 condition 4: the agent can't reach the ledger. It sits in the store dir, which no
// bind mounts; only the store's session/ and claude-projects/ are.
test("the ledger is under no sandbox mount", (t) => {
  for (const legacy of [false, true]) {
    const f = setup(t, { legacy });
    f.collect();
    const file = join(f.store.dir, LEDGER);
    const runDir = join(f.root, "runs", "r1");
    const list = binds({ runDir, workspace: join(f.root, "ws"), adapter: join(f.root, "a.sh"), brokerSocket: join(f.root, "bus.sock"), threadStore: f.store });
    assert.ok(list.some((b) => b.src === f.store.session));
    for (const b of list) {
      if (!b.src || !isAbsolute(b.src)) continue;
      const rel = relative(b.src, file);
      assert.ok(rel.startsWith("..") || isAbsolute(rel), `${b.src} mounts ${file}`);
    }
  }
});