diff --git a/packages/cli/src/launcher.mjs b/packages/cli/src/launcher.mjs index cb9631ddf..b59add550 100644 --- a/packages/cli/src/launcher.mjs +++ b/packages/cli/src/launcher.mjs @@ -306,10 +306,13 @@ export function createLauncher({ if (system?.compactionStreams !== undefined && streams !== system.compactionStreams) { log(`system config: compactionStreams ${JSON.stringify(system.compactionStreams)} isn't a whole number from 32 to 65536; compaction collections open up to the default ${STREAMS} streams`); } - const compactions = (store, harness, { instance, thread, session, run }) => { + // Row 307 (#1814): the route's harness module reads its own store through the kind's + // `compactions`. The call skips call(): call() turns a throw into a launch refusal, and a + // collection that fails must never stop or fail the run, so this catch records it instead. + const compactions = (store, name, harness, { instance, thread, session, run }) => { let c; try { - c = collectCompactions(store, harness, { budget, streams }); + c = collectCompactions(store, { name, compactions: harness.compactions }, { budget, streams }); } catch (e) { log(`run ${run}: compactions: ${e.message}`); c = { count: null, known: false, reason: "store-unreadable" }; @@ -506,7 +509,7 @@ export function createLauncher({ const { run, runDir } = newRun(dataRoot, b.id); const sessionDir = store ? store.session : join(runDir, "session"); if (!store) mkdirSync(sessionDir, { mode: 0o700 }); - if (store) compactions(store, model.harness, { instance, thread, session, run }); + if (store) compactions(store, model.harness, harness, { instance, thread, session, run }); const toolSocket = join(runDir, "tools.sock"); const turnMarker = join(runDir, "turn.done"); const version = harnessVersion !== undefined ? harnessVersion : call(model.harness, "version", () => harness.version({ env })); @@ -647,7 +650,7 @@ export function createLauncher({ } finally { try { unregisterRuntime(dataRoot, managed.id); } catch { log(`run ${run}: processing receipt cleanup refused`); } - if (store) compactions(store, model.harness, { instance, thread, session, run }); + if (store) compactions(store, model.harness, harness, { instance, thread, session, run }); try { lock?.release(); } catch (e) { log(`run ${run}: workspace lock release: ${e.message}`); } children.delete(run); diff --git a/packages/cli/tests/compactions-launch.test.mjs b/packages/cli/tests/compactions-launch.test.mjs index 60afa2c5a..184b1750b 100644 --- a/packages/cli/tests/compactions-launch.test.mjs +++ b/packages/cli/tests/compactions-launch.test.mjs @@ -28,6 +28,7 @@ import { collectCompactions, LEDGER } from "../../seat/src/compactions.mjs"; import { gitEnv } from "../../seat/src/checkout.mjs"; import { sandboxProblem } from "../../seat/src/sandbox.mjs"; import { launchLogFile, SESSION_CREDENTIALS, 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 } from "./helpers.mjs"; @@ -341,7 +342,7 @@ test("REQ-THREAD-2 T3 T4 a clear leaves the retired session's count as it was, i 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, "pi").count, 1); + 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); diff --git a/packages/plugins/README.md b/packages/plugins/README.md index 2b30708b0..e7cc603ad 100644 --- a/packages/plugins/README.md +++ b/packages/plugins/README.md @@ -153,6 +153,21 @@ export returns: `gateTimeout`, `bundle`, `env`, `turnEnd` and `transcript` give its sandbox binds, web, tool names, S0 lines, bundle files, adapter variables, how a turn ends and where its transcripts land. +- `compactions({store, cursors, budget, streams, first})` reads a thread + session's native store for row 270's compaction count (row 307). Core keeps + the ledger (`packages/seat/src/compactions.mjs`) and passes `store` as + `{session, projects}`, the ledger's `cursors` by stream, the read `budget` + and stream cap `streams` from the system config, and `first` when the + store has no ledger yet. The answer is `{found, cursors, known, reason, + version}`: `found` lists `{entry, completedAt, stream}` past the cursors, + `cursors` maps each stream read to `{bytes, tail}` (`tail` is the sha256 of + up to 4096 bytes ending there), `reason` is null, `store-incomplete`, + `store-unreadable`, `not-collected` or `harness-unsupported`, `known` is + true exactly when `reason` is null, and `version` names the store format + read, or null. Core refuses any other answer and writes nothing. pi reads + its streams within the budget and the cap, and answers + `harness-unsupported` on a first collection that finds Claude history. + Claude Code answers `harness-unsupported` with version null (Vikunja 466). - Row 66: both modules wire the unslop check from `tools/unslop-hook`. `mounts` adds its three files read-only and `bundle` names them in the manifest's sources. Pi's `env` loads `extension.ts` as the second entry of diff --git a/packages/plugins/harness/claude-code/index.mjs b/packages/plugins/harness/claude-code/index.mjs index 334d76623..3ae280c99 100644 --- a/packages/plugins/harness/claude-code/index.mjs +++ b/packages/plugins/harness/claude-code/index.mjs @@ -128,3 +128,6 @@ export function env({ files, policy, session }) { // A turn ends when claude exits; there is no marker. export const turnEnd = () => "exit"; export const transcript = ({ runDir, store = null }) => (store ? store.projects : join(runDir, "home", ".claude", "projects")); +// Row 307 (#1814): Claude Code has no pinned compaction format yet (Vikunja 466), so a collection +// reads nothing and the count is unknown with harness-unsupported, never 0. +export const compactions = () => ({ found: [], cursors: {}, known: false, reason: "harness-unsupported", version: null }); diff --git a/packages/plugins/harness/pi/compactions.mjs b/packages/plugins/harness/pi/compactions.mjs new file mode 100644 index 000000000..c6f312c0d --- /dev/null +++ b/packages/plugins/harness/pi/compactions.mjs @@ -0,0 +1,166 @@ +// Row 307 (#1814, Vikunja 598): pi's half of row 270's compaction count (264c, #1775, T1 item 7 +// D6), moved here from packages/seat/src/compactions.mjs. Core keeps the ledger and calls +// `compactions` through the harness kind (packages/plugins/README.md has the contract). This file +// reads only the session's native pi store and the Claude history beside it. +// +// Pi (pinned 0.85.1): a completed compaction is one `type: "compaction"` entry, +// written by SessionManager.appendCompaction after the summary exists. A start +// or an aborted compaction writes none. A `compactionSummary` message, a +// message inside a compaction's retainedTail and a branch_summary are not +// compactions. Pi's createBranchedSession and forkFrom copy entries into a new +// file of the same directory with their id and timestamp unchanged, so a +// compaction's key is (entry id, entry timestamp) over the whole store, and the +// stream it was first seen in is recorded beside it. +// +// A session whose first collection finds Claude history in claude-projects/ +// from before the ledger existed is unknown with harness-unsupported (row 270 +// r2, B3): those runs' compactions can't be counted (Vikunja 466). + +import { createHash } from "node:crypto"; +import { closeSync, constants as fsc, fstatSync, lstatSync, openSync, readdirSync, readSync } from "node:fs"; +import { join } from "node:path"; + +export const PI_VERSION = "0.85.1"; +const TAIL = 4096; +const CHUNK = 1 << 20; +const STREAM = /^[A-Za-z0-9._-]{1,200}\.jsonl$/; +// store-incomplete outranks store-unreadable: lost entries don't come back. +// not-collected is last: the next collection reads on. +const REASONS = ["store-incomplete", "store-unreadable", "not-collected"]; + +const digest = (buf) => createHash("sha256").update(buf).digest("hex"); + +// Only the short tail reads (at most TAIL bytes) use this. +function readRange(fd, from, to) { + const out = Buffer.alloc(to - from); + let off = 0; + while (off < out.length) { + const n = readSync(fd, out, off, Math.min(CHUNK, out.length - off), from + off); + if (n === 0) break; + off += n; + } + return out.subarray(0, off); +} + +// One collection over a Pi session directory. `cursors` are the ledger's, +// by stream. Opens at most `streams` streams, in name order, and reads at +// most `budget` bytes past their cursors, over all of them. +// Returns the compactions found past each cursor, the new cursors, and the +// problems: store-unreadable (a file that can't be read or isn't a Pi +// session), store-incomplete (a stream that shrank, changed or went away, or +// a line that isn't JSON) and not-collected (the budget or the stream cap +// ran out first). A +// missing directory is empty. +function scanPi(dir, cursors, budget, streams) { + const found = [], next = {}, problems = new Set(); + let left = budget; + let names; + try { + names = readdirSync(dir); + } catch (e) { + if (e.code !== "ENOENT") problems.add("store-unreadable"); + names = []; + } + const present = new Set(names); + for (const stream of Object.keys(cursors)) if (!present.has(stream)) problems.add("store-incomplete"); + const visit = names.filter((n) => n.endsWith(".jsonl")).sort(); + if (visit.length > streams) problems.add("not-collected"); + for (const name of visit.slice(0, streams)) { + const path = join(dir, name); + let fd; + try { + if (!STREAM.test(name) || !lstatSync(path).isFile()) throw new Error("not a regular stream file"); + fd = openSync(path, fsc.O_RDONLY | fsc.O_NOFOLLOW); + } catch { + problems.add("store-unreadable"); + continue; + } + try { + const size = fstatSync(fd).size; + const c = cursors[name]; + let from = 0; + if (c) { + const tail = c.bytes > size ? null : readRange(fd, Math.max(0, c.bytes - TAIL), c.bytes); + if (tail && tail.length === Math.min(TAIL, c.bytes) && digest(tail) === c.tail) from = c.bytes; + else problems.add("store-incomplete"); + } + let header = from > 0; + const take = (raw) => { + const line = raw.toString("utf8"); + if (!line.trim()) return; + let e; + try { + e = JSON.parse(line); + } catch { + e = null; + } + if (!e || typeof e !== "object" || Array.isArray(e)) { + problems.add("store-incomplete"); + return; + } + if (!header) { + header = true; + if (e.type !== "session") problems.add("store-unreadable"); + return; + } + if (e.type === "compaction" && typeof e.id === "string" && e.id) { + found.push({ entry: e.id, completedAt: typeof e.timestamp === "string" ? e.timestamp : null, stream: name }); + } + }; + // A CHUNK at a time, up to the budget left. Complete lines only: a line + // split across chunks waits in `carry`, and a partial last line waits + // for its newline. `end` is past the last whole line. + const stop = Math.min(size, from + left); + const chunk = Buffer.alloc(Math.min(CHUNK, stop - from)); + let pos = from, end = from, carry = []; + while (pos < stop) { + const n = readSync(fd, chunk, 0, Math.min(chunk.length, stop - pos), pos); + if (n === 0) break; + const view = chunk.subarray(0, n); + let i = 0; + for (let nl = view.indexOf(0x0a); nl !== -1; nl = view.indexOf(0x0a, i)) { + take(carry.length ? Buffer.concat([...carry, view.subarray(i, nl)]) : view.subarray(i, nl)); + carry = []; + i = nl + 1; + end = pos + i; + } + if (i < n) carry.push(Buffer.from(view.subarray(i))); + pos += n; + } + left -= pos - from; + if (stop < size) problems.add("not-collected"); + next[name] = { bytes: end, tail: digest(readRange(fd, Math.max(0, end - TAIL), end)) }; + } catch { + problems.add("store-unreadable"); + } finally { + closeSync(fd); + } + } + return { found, cursors: next, problems: [...problems].sort() }; +} + +// Row 270 r2 (B3): whether the store's claude-projects/ holds anything. A +// directory that can't be read throws, so the collection is not recorded and +// the next one checks again. +function claudeHistory(store) { + if (!store.projects) return false; + try { + return readdirSync(store.projects).length > 0; + } catch (e) { + if (e.code === "ENOENT") return false; + throw e; + } +} + +// The harness kind's `compactions` for pi: the compactions past the ledger's +// cursors, the moved cursors, and the reason the count isn't whole, if any. +export function compactions({ store, cursors, budget, streams, first }) { + if (first && claudeHistory(store)) { + // Claude runs before this ledger: their compactions can't be counted + // (Vikunja 466), so the count is unknown for good, as after a Claude run. + return { found: [], cursors: {}, known: false, reason: "harness-unsupported", version: PI_VERSION }; + } + const scan = scanPi(store.session, cursors, budget, streams); + const reason = REASONS.find((p) => scan.problems.includes(p)) ?? null; + return { found: scan.found, cursors: scan.cursors, known: reason === null, reason, version: PI_VERSION }; +} diff --git a/packages/plugins/harness/pi/index.mjs b/packages/plugins/harness/pi/index.mjs index 068999ec0..2a3b2037f 100644 --- a/packages/plugins/harness/pi/index.mjs +++ b/packages/plugins/harness/pi/index.mjs @@ -73,3 +73,5 @@ export function env({ policy, tools, session }) { // The extension writes the turn marker at agent_end; an exit without it is a failed turn. export const turnEnd = () => "marker"; export const transcript = ({ runDir, store = null }) => (store ? store.session : join(runDir, "session")); +// Row 307 (#1814): the session store's compactions, read from its pi streams (compactions.mjs). +export { compactions } from "./compactions.mjs"; diff --git a/packages/plugins/src/kinds.mjs b/packages/plugins/src/kinds.mjs index da289d86e..b16f19193 100644 --- a/packages/plugins/src/kinds.mjs +++ b/packages/plugins/src/kinds.mjs @@ -49,10 +49,12 @@ export const KINDS = Object.freeze({ // env({ files, policy, tools, session }) -> its adapter variables, written into session.json // turnEnd() -> "marker" (a turn ends only with the turn marker) | "exit" // transcript({ runDir, store }) -> the directory its transcripts land in, for usage + // compactions({ store, cursors, budget, streams, first }) -> { found, cursors, known, reason, version }, + // the session store's compactions past the ledger's cursors (row 307) harness: Object.freeze({ kind: "harness", interfaceVersion: 1, switch: "list", exports: Object.freeze(["adapter", "levels", "preflight", "version", "countsAccount", "mounts", "builtinWeb", "builtins", - "typedName", "reliesOn", "gateTimeout", "bundle", "env", "turnEnd", "transcript"]), + "typedName", "reliesOn", "gateTimeout", "bundle", "env", "turnEnd", "transcript", "compactions"]), }), }); diff --git a/packages/plugins/tests/kinds.test.mjs b/packages/plugins/tests/kinds.test.mjs index a9d03f79e..8601023c4 100644 --- a/packages/plugins/tests/kinds.test.mjs +++ b/packages/plugins/tests/kinds.test.mjs @@ -17,7 +17,7 @@ test("REQ-PLUG-1 REQ-LIMIT-4 the registry: usage-source at interface version 1, assert.deepEqual({ ...KINDS.harness, exports: [...KINDS.harness.exports] }, { kind: "harness", interfaceVersion: 1, switch: "list", exports: ["adapter", "levels", "preflight", "version", "countsAccount", "mounts", "builtinWeb", "builtins", - "typedName", "reliesOn", "gateTimeout", "bundle", "env", "turnEnd", "transcript"], + "typedName", "reliesOn", "gateTimeout", "bundle", "env", "turnEnd", "transcript", "compactions"], }); assert.ok(Object.isFrozen(KINDS.harness) && Object.isFrozen(KINDS.harness.exports)); assert.ok(Object.isFrozen(KINDS) && Object.isFrozen(KINDS["usage-source"]) && Object.isFrozen(KINDS["limit-check"].exports)); diff --git a/packages/seat/src/compactions.mjs b/packages/seat/src/compactions.mjs index a30a26f0a..5c20f261c 100644 --- a/packages/seat/src/compactions.mjs +++ b/packages/seat/src/compactions.mjs @@ -9,19 +9,14 @@ // workspace lock (session.mjs lockSessionWorkspace); each collection is one // O_APPEND write. The count is rebuilt from the ledger, never from a page. // -// Pi (pinned 0.85.1): a completed compaction is one `type: "compaction"` entry, -// written by SessionManager.appendCompaction after the summary exists. A start -// or an aborted compaction writes none. A `compactionSummary` message, a -// message inside a compaction's retainedTail and a branch_summary are not -// compactions. Pi's createBranchedSession and forkFrom copy entries into a new -// file of the same directory with their id and timestamp unchanged, so a -// compaction's key is (entry id, entry timestamp) over the whole store, and the -// stream it was first seen in is recorded beside it. -// -// Claude Code has no pinned compaction format yet (Vikunja 466): a session -// any Claude run collected for is unknown with harness-unsupported, never 0. -// So is one whose first collection finds Claude history in claude-projects/ -// from before this ledger existed (row 270 r2, B3). +// What a compaction is depends on the harness, so core never reads a native +// stream itself (row 307, #1814, Vikunja 598). Each collection calls the route's +// harness module's `compactions` (packages/plugins/README.md has the contract), +// checks its answer, and writes the ledger lines from it. A compaction's key is +// (entry, completedAt) over the whole store, and the stream it was first seen in +// is recorded beside it. A harness with no pinned compaction format answers +// harness-unsupported, and so does one that can't count the store's history: a +// session any such collection ran for is unknown for good, never 0. // // One collection reads at most a budget of stream bytes (row 270 r2, B1). // The budget is the system config's compactionReadBytes, which only a human @@ -46,14 +41,11 @@ // that shrank, changed under its cursor or went away is store-incomplete from // then on: entries may be lost, so the count stays a lower bound for good. -import { createHash } from "node:crypto"; -import { closeSync, constants as fsc, fstatSync, lstatSync, openSync, readdirSync, readFileSync, readSync, writeSync } from "node:fs"; +import { closeSync, constants as fsc, fstatSync, lstatSync, openSync, readFileSync, writeSync } from "node:fs"; import { join } from "node:path"; import { ID_PATTERN } from "../../business/src/vocabulary.mjs"; -export const PI_VERSION = "0.85.1"; export const LEDGER = "compactions.jsonl"; -const TAIL = 4096; const CHUNK = 1 << 20; // BUILD.md (queue-270, round 2) has the measured stream sizes behind the default. export const READ_BYTES = 64 * CHUNK; @@ -61,9 +53,7 @@ const READ_MIN = CHUNK, READ_MAX = 1024 * CHUNK; // BUILD.md (queue-270, round 3) has the measured stream counts behind the default. export const STREAMS = 1024; const STREAMS_MIN = 32, STREAMS_MAX = 65536; -const STREAM = /^[A-Za-z0-9._-]{1,200}\.jsonl$/; -const digest = (buf) => createHash("sha256").update(buf).digest("hex"); const keyOf = (entry, completedAt) => JSON.stringify([entry, completedAt]); const segment = (x) => typeof x === "string" && ID_PATTERN.test(x) && !x.includes(".."); @@ -87,115 +77,6 @@ export function readStreams(value) { return Number.isSafeInteger(value) && value >= STREAMS_MIN && value <= STREAMS_MAX ? value : STREAMS; } -// Only the short tail reads (at most TAIL bytes) use this. -function readRange(fd, from, to) { - const out = Buffer.alloc(to - from); - let off = 0; - while (off < out.length) { - const n = readSync(fd, out, off, Math.min(CHUNK, out.length - off), from + off); - if (n === 0) break; - off += n; - } - return out.subarray(0, off); -} - -// One collection over a Pi session directory. `cursors` are the ledger's, -// by stream. Opens at most `streams` streams, in name order, and reads at -// most `budget` bytes past their cursors, over all of them. -// Returns the compactions found past each cursor, the new cursors, and the -// problems: store-unreadable (a file that can't be read or isn't a Pi -// session), store-incomplete (a stream that shrank, changed or went away, or -// a line that isn't JSON) and not-collected (the budget or the stream cap -// ran out first). A -// missing directory is empty. -export function scanPi(dir, cursors = {}, budget = READ_BYTES, streams = STREAMS) { - const found = [], next = {}, problems = new Set(); - let left = budget; - let names; - try { - names = readdirSync(dir); - } catch (e) { - if (e.code !== "ENOENT") problems.add("store-unreadable"); - names = []; - } - const present = new Set(names); - for (const stream of Object.keys(cursors)) if (!present.has(stream)) problems.add("store-incomplete"); - const visit = names.filter((n) => n.endsWith(".jsonl")).sort(); - if (visit.length > streams) problems.add("not-collected"); - for (const name of visit.slice(0, streams)) { - const path = join(dir, name); - let fd; - try { - if (!STREAM.test(name) || !lstatSync(path).isFile()) throw new Error("not a regular stream file"); - fd = openSync(path, fsc.O_RDONLY | fsc.O_NOFOLLOW); - } catch { - problems.add("store-unreadable"); - continue; - } - try { - const size = fstatSync(fd).size; - const c = cursors[name]; - let from = 0; - if (c) { - const tail = c.bytes > size ? null : readRange(fd, Math.max(0, c.bytes - TAIL), c.bytes); - if (tail && tail.length === Math.min(TAIL, c.bytes) && digest(tail) === c.tail) from = c.bytes; - else problems.add("store-incomplete"); - } - let header = from > 0; - const take = (raw) => { - const line = raw.toString("utf8"); - if (!line.trim()) return; - let e; - try { - e = JSON.parse(line); - } catch { - e = null; - } - if (!e || typeof e !== "object" || Array.isArray(e)) { - problems.add("store-incomplete"); - return; - } - if (!header) { - header = true; - if (e.type !== "session") problems.add("store-unreadable"); - return; - } - if (e.type === "compaction" && typeof e.id === "string" && e.id) { - found.push({ entry: e.id, completedAt: typeof e.timestamp === "string" ? e.timestamp : null, stream: name }); - } - }; - // A CHUNK at a time, up to the budget left. Complete lines only: a line - // split across chunks waits in `carry`, and a partial last line waits - // for its newline. `end` is past the last whole line. - const stop = Math.min(size, from + left); - const chunk = Buffer.alloc(Math.min(CHUNK, stop - from)); - let pos = from, end = from, carry = []; - while (pos < stop) { - const n = readSync(fd, chunk, 0, Math.min(chunk.length, stop - pos), pos); - if (n === 0) break; - const view = chunk.subarray(0, n); - let i = 0; - for (let nl = view.indexOf(0x0a); nl !== -1; nl = view.indexOf(0x0a, i)) { - take(carry.length ? Buffer.concat([...carry, view.subarray(i, nl)]) : view.subarray(i, nl)); - carry = []; - i = nl + 1; - end = pos + i; - } - if (i < n) carry.push(Buffer.from(view.subarray(i))); - pos += n; - } - left -= pos - from; - if (stop < size) problems.add("not-collected"); - next[name] = { bytes: end, tail: digest(readRange(fd, Math.max(0, end - TAIL), end)) }; - } catch { - problems.add("store-unreadable"); - } finally { - closeSync(fd); - } - } - return { found, cursors: next, problems: [...problems].sort() }; -} - // The ledger as recorded. Throws on a ledger that can't be read. `bad` is set // by any line that isn't a v 1 record, a torn last line included. export function readLedger(dir) { @@ -229,28 +110,16 @@ export function readLedger(dir) { else if (r.kind === "collected") { last = r; if (r.reason === "store-incomplete") incomplete = true; - if (r.harness !== "pi" || r.reason === "harness-unsupported") unsupported = true; + if (r.reason === "harness-unsupported") unsupported = true; } else bad = true; } return { events, cursors, last, incomplete, unsupported, bad, torn }; } -// Row 270 r2 (B3): whether the store's claude-projects/ holds anything. A -// directory that can't be read throws, so the collection is not recorded and -// the next one checks again. -function claudeHistory(store) { - if (!store.projects) return false; - try { - return readdirSync(store.projects).length > 0; - } catch (e) { - if (e.code === "ENOENT") return false; - throw e; - } -} - // What session.list shows: { count, known, reason, at }. Known only when the -// ledger is whole, every collection ran on Pi and none lost entries. Any other -// answer is unknown with a reason and a lower bound, or null for none: never 0. +// ledger is whole, no collection was harness-unsupported and none lost +// entries. Any other answer is unknown with a reason and a lower bound, or null +// for none: never 0. export function compactionCount(dir) { const unknown = (reason, count = null, at = null) => ({ count: count || null, known: false, reason, at }); let ledger; @@ -270,39 +139,57 @@ export function compactionCount(dir) { return { count, known: true, reason: null, at }; } -// One collection for a session's store ({ dir, session } from sessionStore), -// at run start (resume) and run end, under the session's workspace lock. -// Appends the new compactions, the moved cursors and one collected line in a -// single O_APPEND write, then answers the count. A key already in the ledger -// is never written again, and nothing rewrites the file. +const REASONS = ["store-incomplete", "store-unreadable", "not-collected", "harness-unsupported"]; +const HEX = /^[0-9a-f]{64}$/; + +// The harness module's answer, checked before anything is written: what the +// export's contract allows and nothing else. Returns what's wrong, or null. +function checkAnswer(a) { + const plain = (x) => x !== null && typeof x === "object" && !Array.isArray(x); + if (!plain(a)) return "not an object"; + if (!Array.isArray(a.found)) return "found isn't a list"; + for (const f of a.found) { + if (!plain(f) || typeof f.entry !== "string" || !f.entry) return "a found entry has no entry id"; + if (f.completedAt !== null && typeof f.completedAt !== "string") return `found ${f.entry} has a completedAt that isn't a string or null`; + if (typeof f.stream !== "string" || !f.stream) return `found ${f.entry} has no stream`; + } + if (!plain(a.cursors)) return "cursors isn't an object"; + for (const [stream, c] of Object.entries(a.cursors)) { + if (!plain(c) || !Number.isSafeInteger(c.bytes) || c.bytes < 0 || typeof c.tail !== "string" || !HEX.test(c.tail)) return `cursor ${JSON.stringify(stream)} isn't { bytes, tail }`; + } + if (a.reason !== null && !REASONS.includes(a.reason)) return `reason ${JSON.stringify(a.reason)} isn't one of ${REASONS.join(", ")}`; + if (a.known !== (a.reason === null)) return "known must be true exactly when reason is null"; + if (a.version !== null && typeof a.version !== "string") return "version isn't a string or null"; + return null; +} + +// One collection for a session's store ({ dir, session, projects } from +// sessionStore), at run start (resume) and run end, under the session's +// workspace lock. `harness` is { name, compactions }: the route's harness name +// and its loaded module's export. Appends the new compactions, the moved +// cursors and one collected line in a single O_APPEND write, then answers the +// count. A key already in the ledger is never written again, and nothing +// rewrites the file. A module that throws or answers out of contract throws +// here, before anything is written. export function collectCompactions(store, harness, { now = () => new Date().toISOString(), budget = READ_BYTES, streams = STREAMS } = {}) { const prior = readLedger(store.dir); const ledger = prior ?? { events: new Map(), cursors: {}, torn: false }; const at = now(); + const answer = harness.compactions({ store: { session: store.session, projects: store.projects }, cursors: { ...ledger.cursors }, budget, streams, first: prior === null }); + const wrong = checkAnswer(answer); + if (wrong) throw new Error(`harness ${harness.name} compactions: bad answer: ${wrong}`); const out = []; - if (harness === "pi" && !prior && claudeHistory(store)) { - // Claude runs before this ledger: their compactions can't be counted - // (Vikunja 466), so the count is unknown for good, as after a Claude run. - out.push({ v: 1, kind: "collected", at, known: false, reason: "harness-unsupported", harness, version: PI_VERSION }); - } else if (harness === "pi") { - const scan = scanPi(store.session, ledger.cursors, budget, streams); - for (const f of scan.found) { - const key = keyOf(f.entry, f.completedAt); - if (ledger.events.has(key)) continue; - ledger.events.set(key, f); - out.push({ v: 1, kind: "compaction", entry: f.entry, completedAt: f.completedAt, stream: f.stream, at }); - } - for (const [stream, c] of Object.entries(scan.cursors)) { - const old = ledger.cursors[stream]; - if (!old || old.bytes !== c.bytes || old.tail !== c.tail) out.push({ v: 1, kind: "cursor", stream, bytes: c.bytes, tail: c.tail, at }); - } - // store-incomplete outranks store-unreadable: lost entries don't come back. - // not-collected is last: the next collection reads on. - const reason = ["store-incomplete", "store-unreadable", "not-collected"].find((p) => scan.problems.includes(p)) ?? null; - out.push({ v: 1, kind: "collected", at, known: reason === null, reason, harness, version: PI_VERSION }); - } else { - out.push({ v: 1, kind: "collected", at, known: false, reason: "harness-unsupported", harness, version: null }); + for (const f of answer.found) { + const key = keyOf(f.entry, f.completedAt); + if (ledger.events.has(key)) continue; + ledger.events.set(key, f); + out.push({ v: 1, kind: "compaction", entry: f.entry, completedAt: f.completedAt, stream: f.stream, at }); } + for (const [stream, c] of Object.entries(answer.cursors)) { + const old = ledger.cursors[stream]; + if (!old || old.bytes !== c.bytes || old.tail !== c.tail) out.push({ v: 1, kind: "cursor", stream, bytes: c.bytes, tail: c.tail, at }); + } + out.push({ v: 1, kind: "collected", at, known: answer.known, reason: answer.reason, harness: harness.name, version: answer.version }); const fd = openSync(join(store.dir, LEDGER), fsc.O_WRONLY | fsc.O_APPEND | fsc.O_CREAT | fsc.O_NOFOLLOW, 0o600); try { if (!fstatSync(fd).isFile()) throw Object.assign(new Error(`${join(store.dir, LEDGER)} isn't a regular file`), { code: "ledger-unsafe" }); diff --git a/packages/seat/tests/compactions-budget.test.mjs b/packages/seat/tests/compactions-budget.test.mjs index d8025354b..b9c873d7f 100644 --- a/packages/seat/tests/compactions-budget.test.mjs +++ b/packages/seat/tests/compactions-budget.test.mjs @@ -12,6 +12,7 @@ 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; @@ -22,7 +23,7 @@ function setup(t) { 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, harness, { now, ...(budget === undefined ? {} : { budget }), ...(streams === undefined ? {} : { streams }) }); + 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 }; } diff --git a/packages/seat/tests/compactions.test.mjs b/packages/seat/tests/compactions.test.mjs index 5df66084a..5b3ea6b8b 100644 --- a/packages/seat/tests/compactions.test.mjs +++ b/packages/seat/tests/compactions.test.mjs @@ -11,6 +11,8 @@ 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-")); @@ -18,7 +20,7 @@ function setup(t, { legacy = false } = {}) { 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, harness, { now }); + 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 };