gizmoduck: row 307 (#1814, Vikunja 598) harness kind gains compactions: pi's stream scan and Claude-history check move to packages/plugins/harness/pi/compactions.mjs, claude-code answers harness-unsupported, core keeps the ledger and checks each answer, readLedger reads the reason only, the launcher passes the loaded module; four old-test edits under mos-1459

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);

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);

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

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

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

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";

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"]),
   }),
 });
 

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));

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" });

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

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