// 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 }; }