stack/packages/plugins/harness/pi/compactions.mjs

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