233 lines
13 KiB
JavaScript
233 lines
13 KiB
JavaScript
// 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}`);
|
|
}
|
|
}
|
|
});
|