gizmoduck: row 270 (264c, #1775): completed Pi compactions counted into a host ledger beside the session store, keyed (entry id, timestamp) store-wide (sage-2806, sage-2814); launcher collects at run start and end under the workspace lock; session.list rows gain compactions through a runtime-injected reader; not run (slot held)

agents/gizmoduck/work/queue-270/BUILD.md

@@ -42,8 +42,115 @@ No new verb and no new input key. Access is `session.list`'s: the thread owner,
 - What doesn't count: a `message` entry with role `compactionSummary` (a rendered summary,
   docs/session-format.md:154), any message inside a compaction's `retainedTail`, and
   `branch_summary` entries.
-- Identity: the stream is the `*.jsonl` file name under the session's `session/` directory, and the
+- Identity: a compaction is its (entry `id`, entry `timestamp`) across the whole store, not per
-  event is the entry `id`. Pi's ids are unique within a file (`generateId(this.byId)`). An entry
+  stream. Pi's ids are unique only within a file (`generateId(this.byId)`, session-manager.js:21-29),
-  id repeated in one file counts once.
+  and `createBranchedSession` (:1093-1140) and `forkFrom` (:1264-1300) copy entries into a new
+  `*.jsonl` in the same directory with their id and timestamp unchanged. A (stream, id) key would
+  count a compaction made before a branch or fork once per copy. Reported to Sage in
+  gizmoduck-0262; accepted in sage-2814. The ledger line records the stream it was first seen in.
+- Residual (sage-2814): two distinct compactions that share an id and a timestamp count once. That
+  needs an 8-hex collision between two files at the same millisecond. It is the one undercount
+  this key allows, and it never overcounts.
 - Partial trailing line: wait. Only lines that end with a newline are read, as
   packages/conversation/src/pi.mjs:35-38 does.
+
+## Storage: A, a host ledger (sage-2806)
+
+Sage ruled A over B. A keeps the count in `<store dir>/compactions.jsonl`, a file beside the
+mounted `session/` and `claude-projects/`, in the store directory no bind mounts
+(sandbox.mjs:122-127). B would have been a bus table. The tradeoff: A needs no schema step, so 271's
+clear journal stays the only new schema step in the 264 rows, and the count lives with the native
+store it is collected from. The cost is that the bus reads a file on each `session.list`, through a
+reader runtime.mjs injects, because broker.mjs is a RUNNER_FILES file and can't import seat code.
+
+### Ledger
+
+`packages/seat/src/compactions.mjs`. One JSON object per line, `v: 1`:
+
+- `{kind: "compaction", entry, completedAt, stream, at}`: once per (entry, completedAt) key.
+- `{kind: "cursor", stream, bytes, tail, at}`: the stream was read through `bytes`. `tail` is the
+  sha256 of the up to 4096 bytes ending there. Written only when it moved.
+- `{kind: "collected", at, known, reason, harness, version}`: one per collection.
+
+Conditions (sage-2806):
+
+1. Append-only, the launcher its only writer, a key never written twice, nothing rewrites or
+   truncates. The key is the one above, not (stream, id).
+2. Mode 0600, opened `O_WRONLY|O_APPEND|O_CREAT|O_NOFOLLOW`. **Lock used: the session's workspace
+   lock** (`lockSessionWorkspace`, taken in launcher.mjs `start()` and released in the run's exit
+   `finally` or on a start failure). Collection runs at run start, after `check()` gives the
+   harness and before the spawn, and at run end, before the lock is released. Each collection is
+   one `writeSync` loop on one O_APPEND descriptor. A collection error is logged and recorded as a
+   `compactions` launch-log event and never stops or fails the run.
+3. `compactionCount(dir)`:
+
+   | Ledger | Answer |
+   | --- | --- |
+   | missing | `not-collected`, count null |
+   | unreadable, not a regular file, a symlink | `store-unreadable`, count null |
+   | any bad line, a torn last line included | `store-incomplete`, count a lower bound or null |
+   | any collection under a harness other than pi | `harness-unsupported`, lower bound or null |
+   | any collection that found a stream shrunk, changed under its cursor or gone, or a bad stream line | `store-incomplete`, for good |
+   | the last collection hit an unreadable stream or a non-session first line | `store-unreadable` until a clean collection |
+   | otherwise | `known: true`, the exact count |
+
+   Never count 0 when unknown. After a torn line, the next write starts on a fresh line, so the torn
+   line stays one bad line and the new lines stay whole.
+4. `packages/seat/tests/compactions.test.mjs` "the ledger is under no sandbox mount": for a legacy
+   and a non-legacy store, no `binds()` source contains the ledger path.
+5. This section.
+
+### Wiring
+
+- launcher.mjs: `compactions(store, model.harness, ...)` at start (after `newRun`) and in the exit
+  `finally` before `lock?.release()`. Thread sessions only.
+- broker.mjs: a `compactions` constructor option, TypeError-checked like `access`. `session.list`
+  selects `op` to tell a legacy session (`legacy-3l`) and adds `compactions` to each row; `op` is
+  not returned. No reader gives `not-collected`; a reader that throws gives `store-unreadable`.
+  No import added.
+- runtime.mjs: passes `({business, thread, session, legacy}) => compactionCount(storeDir(...))`.
+- packages/bus/README.md: the `session.list` row.
+
+Huey's 272 broker hunk (reply citation predicate, near broker.mjs:5167 on huey/row272) doesn't
+touch `session.list`.
+
+## Tests
+
+New files only, so no pre-existing test changes:
+
+- packages/seat/tests/compactions.test.mjs: zero, a failed start and non-compaction entries; one and
+  three; a compaction copied into a branched stream counts once (sage-2814); one copy in two
+  streams in one collection; the same id at another time counts twice; resume with cursors and
+  keys written once; ledger mode 0600; a partial trailing stream line waits; a missing, a directory
+  and a symlinked ledger; a torn and a bad ledger line; shrink, rewrite and remove sticky; a
+  symlinked stream and a non-header first line; Claude harness-unsupported and sticky; `storeDir`
+  against `sessionStore`; the mount test.
+- packages/bus/tests/session-compactions.test.mjs: the key per row with the store identity
+  asked for, legacy true for `legacy-3l`, no reader and a throwing reader, the TypeError.
+
+Run limits (mos-1275): each file whole with `timeout -k 10 120`, TMPDIR from
+`mktemp -d /mnt/storage/scratch/tmp/gXXX`. Then the touched files' suites: seat session.test,
+sandbox.test; cli thread-launch, session-launch; bus sessions, project-runtime, tester-writes.
+
+## Mutants
+
+Each must turn a new test red:
+
+- M1 key per stream at write: `keyOf(f.entry, f.completedAt)` → `keyOf(f.stream + f.entry, ...)` in
+  `collectCompactions`. The reader still dedupes, so the count holds; the kill is the
+  branched-copy test's ledger lines (`[["c1","a.jsonl"],["c2","b.jsonl"]]`) and the two-stream
+  test's one line.
+- M2 key without the timestamp: `keyOf(entry, null)` everywhere. Kill: same-id-another-time test.
+- M3 count `compactionSummary` messages: count `e.message?.role === "compactionSummary"` too. Kill:
+  the zero test.
+- M4 ignore the cursor tail: `digest(tail) === c.tail` → `true`. Kill: rewrite subtest.
+- M5 torn line skipped: `bad = torn = true` → `torn = true`. Kill: torn-line test.
+- M6 no newline after a torn line: drop the `ledger.torn ? "\n"` prefix. Kill: torn-line test's bad
+  lines.
+- M7 incomplete not sticky: drop `if (ledger.incomplete)`. Kill: the "later clean collection"
+  assertion.
+- M8 harness not sticky: drop `if (ledger.unsupported)`. Kill: Claude then Pi test.
+- M9 0 when unknown: `count: count || null` → `count`. Kill: torn-line test's null.
+- M10 removed stream ignored: drop the missing-cursor loop. Kill: remove subtest.
+- M11 broker legacy flag: `op === 'legacy-3l'` → `false`. Kill: bus legacy test.
+- M12 broker default 0: no reader → `{count: 0, known: true, ...}`. Kill: bus no-reader test.

packages/bus/README.md

@@ -246,7 +246,7 @@ a receipt with empty `changes`. A refusal is `action.refused` with its code alon
 | `thread.messages` | `{thread,session?,after?,limit?}`; human sessions only. The session's messages oldest first, after the message `after` names (`message-not-found` when it isn't in the session), at most `limit` (default 50, 1 to 200). A thread shared with the caller reads too, any session of it (row 206). Writes nothing. |
 | `thread.share`, `thread.unshare` | `{thread,to}`; the thread's owner only (row 206, #1704). `to` is `user:<id>`, the business human's name or `*`; yourself refuses `invalid-request`. A share refuses `share-exists` when one is in force, and `share-not-covered` when a `user:` target has no live grant covering the thread. An unshare refuses `share-not-found` when none is. Each writes one `thread_shares` row. Returns `{thread,to,op}`. |
 | `thread.shares` | `{thread}`; the thread's owner only. The shares in force, `[{to,at,by}]`, oldest first. Writes nothing. |
-| `session.list` | `{thread}`; human and tester sessions (row 264a, #1769). The thread's sessions oldest first, `[{id,at,name,predecessor,default}]`. Writes nothing. |
+| `session.list` | `{thread}`; human and tester sessions (row 264a, #1769). The thread's sessions oldest first, `[{id,at,name,predecessor,default,compactions}]`. `compactions` is `{count,known,reason,at}` from the host's ledger (row 270, #1775): `known` true gives the exact count of completed Pi compactions; otherwise `count` is a lower bound or null and `reason` is `harness-unsupported`, `not-collected`, `store-unreadable` or `store-incomplete`. Writes nothing. |
 | `session.create` | `{thread,name,op}`; human and tester sessions. A new session of the thread, metadata only. A retry with the same `op` and name returns the first row; another name under that `op` refuses `session-op-conflict`, and a taken name `session-exists`. Returns `{id,thread,name}`. |
 
 ### Projects and threads (row 121, #1615)

packages/bus/src/broker.mjs

@@ -394,6 +394,7 @@ export class Broker {
   // FED R3: () => the checked access document (R1's loadAccess), read once per user request. Null,
   // a throw or no document leaves every user with no grant (design 3.5, fail closed).
   #access;
+  #compactions;
   // r6 X3: interrupted host grants whose remote undo waits for reconcileHosts, each with its claim.
   #hostUndo = [];
   // r6 section 6 (N4): revokes a restart cut short, whose step 3 waits for reconcileHosts.
@@ -419,13 +420,14 @@ export class Broker {
   #clock = 0;
   #processing;
   // credentialStatus is the runtime's metadata-only Credentials.status(); the broker never gets use().
-  constructor({ store, businesses, secretCheck = () => {}, credentialStatus = () => [], organization = null, project = null, library = null, usage = null, limits = null, harnessOn = null, credentialVerbs = null, hostGrant = null, hostKeys = null, processing = null, testerOn = null, roleFile = null, giteaGrant = null, giteaKeys = null, access = null }) {
+  constructor({ store, businesses, secretCheck = () => {}, credentialStatus = () => [], organization = null, project = null, library = null, usage = null, limits = null, harnessOn = null, credentialVerbs = null, hostGrant = null, hostKeys = null, processing = null, testerOn = null, roleFile = null, giteaGrant = null, giteaKeys = null, access = null, compactions = null }) {
     if (!object(businesses) || !Object.keys(businesses).length) fail('invalid-business');
     // Row 194 b15 (sage-2501): the function-argument checks come before anything reads the store.
     if (harnessOn !== null && typeof harnessOn !== 'function') throw new TypeError('harnessOn must be a function');
     if (testerOn !== null && typeof testerOn !== 'function') throw new TypeError('testerOn must be a function');
     if (roleFile !== null && typeof roleFile !== 'function') throw new TypeError('roleFile must be a function');
     if (access !== null && typeof access !== 'function') throw new TypeError('access must be a function');
+    if (compactions !== null && typeof compactions !== 'function') throw new TypeError('compactions must be a function');
     this.#businesses = structuredClone(businesses);
     this.#store = store;
     // A Broker is a trusted host startup, including in-process hosts. Never repair an existing state as off.
@@ -445,6 +447,7 @@ export class Broker {
     this.#giteaGrant = giteaGrant;
     this.#giteaKeys = giteaKeys;
     this.#access = access;
+    this.#compactions = compactions;
     this.#project = project;
     this.#library = library;
     this.#usage = usage;
@@ -763,6 +766,15 @@ export class Broker {
     if (s.tester && this.#store.get('SELECT owner FROM threads WHERE id=?', thread).owner !== this.#owner(s)) fail('thread-not-owned');
     return thread;
   }
+  #compactionCount(t, session, legacy) {
+    const unknown = (reason) => ({ count: null, known: false, reason, at: null });
+    if (!this.#compactions) return unknown('not-collected');
+    try {
+      return this.#compactions({ business: t.business, thread: t.id, session, legacy });
+    } catch {
+      return unknown('store-unreadable');
+    }
+  }
   #thread(s, id) {
     const t = this.#threadRow(s, id);
     // REQ-USER-2: a user reaches only their own threads.
@@ -5939,14 +5951,17 @@ export class Broker {
       }
       // Row 264a (#1769, T1 item 7): a thread's sessions, oldest first, and which one is the default.
       // Each user reaches only their own threads' sessions (REQ-USER-2); there are no shares (Q7).
+      // Row 270 (264c, D6): each row's compaction count, from the host's ledger reader (the bus
+      // host's seat compactions.mjs; this file is mounted in the sandbox and can't import it).
+      // Without a reader the count is unknown, never 0.
       case 'session.list': {
         if (!s.tester) this.#human(s);
         keys(a, ['thread'], ['thread']);
         const t = this.#thread(s, a.thread);
         const fallback = this.#defaultSession(t.id);
         return this.#store
-          .all('SELECT id,at,name,predecessor FROM sessions WHERE thread=? ORDER BY seq', t.id)
+          .all('SELECT id,at,name,predecessor,op FROM sessions WHERE thread=? ORDER BY seq', t.id)
-          .map((r) => ({ ...r, default: r.id === fallback }));
+          .map(({ op, ...r }) => ({ ...r, default: r.id === fallback, compactions: this.#compactionCount(t, r.id, op === 'legacy-3l') }));
       }
       // D8: a new session is metadata only. Its run starts when the launcher starts it (264b).
       // The op key makes a retry find the first row: the same op and name answer it, another name

packages/bus/src/runtime.mjs

@@ -9,6 +9,7 @@ import { Credentials } from './credentials.mjs';
 import { serve } from './server.mjs';
 import { launchSettings } from '../../business/src/launch-settings.mjs';
 import { readSessions } from '../../seat/src/session.mjs';
+import { compactionCount, storeDir } from '../../seat/src/compactions.mjs';
 import { publishRemote, workingRepo, WorkingError } from '../../seat/src/working.mjs';
 import { createLoader, hasModule } from '../../plugins/src/loader.mjs';
 import { pluginSettings } from '../../business/src/plugin-settings.mjs';
@@ -261,6 +262,11 @@ export async function startBroker({
     broker = new Broker({
       // FED R3: the access file is read per user request; a refusal leaves the user no grant.
       access: access ? () => loadAccess(access.file, access.context) : null,
+      // Row 270 (264c): session.list's compaction count, read from the session store's ledger.
+      compactions: ({ business, thread, session, legacy }) => {
+        const dir = storeDir(dataRoot, business, thread, session, { legacy });
+        return dir ? compactionCount(dir) : { count: null, known: false, reason: 'store-unreadable', at: null };
+      },
       hostGrant: runRoot ? hostGrant({ configDir: organization.dir, runRoot }) : null,
       // Row 235 (#1741, r6 X4, X5): the grant executor's claim, register and issue, never use().
       hostKeys: storeRoot ? grantKeys({ credentials, storeRoot }) : null,

packages/bus/tests/session-compactions.test.mjs

@@ -0,0 +1,87 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { mkdtempSync, rmSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { Store } from '../src/store.mjs';
+import { Broker } from '../src/broker.mjs';
+
+// Row 270 (264c, #1775, T1 item 7 D6): session.list's `compactions` key, from the host's ledger
+// reader (runtime.mjs gives the seat's compactionCount). The reader itself is
+// packages/seat/tests/compactions.test.mjs's.
+const businesses = {
+  demo: {
+    id: 'demo',
+    human: 'jason',
+    arbiters: { technical: 'pm', delivery: 'chief' },
+    roles: {
+      chief: { definition: 'chief-of-staff', authority: { withinRole: ['message.send'], crossRole: [] } },
+      pm: { definition: 'pm', authority: { withinRole: ['message.send'], crossRole: [] } },
+    },
+    definitions: { 'chief-of-staff': {}, pm: {} },
+    projects: { stack: { root: '/nonexistent' } },
+  },
+};
+
+function setup(t, options = {}) {
+  const root = mkdtempSync(join(tmpdir(), 'bus-compactions-'));
+  const store = new Store(root);
+  t.after(() => {
+    store.close();
+    rmSync(root, { recursive: true, force: true });
+  });
+  const b = new Broker({ store, businesses, ...options });
+  const human = b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
+  const call = (verb, args = {}) => b.request(human, { verb, args });
+  const project = call('project.create', { name: 'stack work', repo: 'stack' }).id;
+  const A = call('thread.create', { project, name: 'alpha' }).id;
+  return { store, call, A };
+}
+
+test('session.list rows carry the reader count for each session, with its store identity', (t) => {
+  const asked = [];
+  const f = setup(t, {
+    compactions: (q) => {
+      asked.push(q);
+      return { count: asked.length, known: true, reason: null, at: '2026-10-22T12:00:00.000Z' };
+    },
+  });
+  const S = f.call('session.create', { thread: f.A, name: 'second', op: 'op1' }).id;
+  const rows = f.call('session.list', { thread: f.A });
+  assert.deepEqual(rows.map((r) => [r.id, r.compactions]), [
+    [rows[0].id, { count: 1, known: true, reason: null, at: '2026-10-22T12:00:00.000Z' }],
+    [S, { count: 2, known: true, reason: null, at: '2026-10-22T12:00:00.000Z' }],
+  ]);
+  assert.deepEqual(asked, rows.map((r) => ({ business: 'demo', thread: f.A, session: r.id, legacy: false })));
+  assert.ok(rows.every((r) => !('op' in r)));
+});
+
+// A legacy session (op legacy-3l, from the 3k-3l migration) keeps the thread's store.
+test('a legacy session is asked for with legacy true', (t) => {
+  const asked = [];
+  const f = setup(t, { compactions: (q) => (asked.push(q), { count: 0, known: true, reason: null, at: null }) });
+  f.store.run("UPDATE sessions SET op='legacy-3l' WHERE thread=?", f.A);
+  f.call('session.list', { thread: f.A });
+  assert.deepEqual(asked.map((q) => q.legacy), [true]);
+});
+
+test('without a reader the count is unknown, never 0; a reader that throws is store-unreadable', (t) => {
+  const f = setup(t);
+  assert.deepEqual(f.call('session.list', { thread: f.A }).map((r) => r.compactions), [{ count: null, known: false, reason: 'not-collected', at: null }]);
+  const g = setup(t, {
+    compactions: () => {
+      throw new Error('EACCES');
+    },
+  });
+  assert.deepEqual(g.call('session.list', { thread: g.A }).map((r) => r.compactions), [{ count: null, known: false, reason: 'store-unreadable', at: null }]);
+});
+
+test('compactions must be a function', (t) => {
+  const root = mkdtempSync(join(tmpdir(), 'bus-compactions-'));
+  const store = new Store(root);
+  t.after(() => {
+    store.close();
+    rmSync(root, { recursive: true, force: true });
+  });
+  assert.throws(() => new Broker({ store, businesses, compactions: {} }), { name: 'TypeError', message: 'compactions must be a function' });
+});

packages/cli/src/launcher.mjs

@@ -162,6 +162,7 @@ import {
   writeSessionFile,
 } from "../../seat/src/session.mjs";
 import { BWRAP, sandboxProblem } from "../../seat/src/sandbox.mjs";
+import { collectCompactions } from "../../seat/src/compactions.mjs";
 import { CheckoutError, ensureCheckouts } from "../../seat/src/checkout.mjs";
 import { createGitSocket } from "./git-socket.mjs";
 import { cmdlineOf, startTimeOf } from "../../seat/src/proc.mjs";
@@ -283,6 +284,22 @@ export function createLauncher({
     }
   };
 
+  // Row 270 (264c, D6): count a thread session's completed compactions into its
+  // store's ledger, at run start (what earlier runs and a resume left) and at
+  // run end. The caller holds the session's workspace lock. A failure is
+  // logged and recorded, and never stops or fails the run: the count then
+  // reads unknown with its reason.
+  const compactions = (store, harness, { instance, thread, session, run }) => {
+    let c;
+    try {
+      c = collectCompactions(store, harness);
+    } catch (e) {
+      log(`run ${run}: compactions: ${e.message}`);
+      c = { count: null, known: false, reason: "store-unreadable" };
+    }
+    record({ event: "compactions", instance, thread, session, run, count: c.count, known: c.known, reason: c.reason });
+  };
+
   function serialize(fn) {
     const run = chain.then(fn);
     chain = run.catch(() => {});
@@ -472,6 +489,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 });
     const toolSocket = join(runDir, "tools.sock");
     const turnMarker = join(runDir, "turn.done");
     const version = harnessVersion !== undefined ? harnessVersion : call(model.harness, "version", () => harness.version({ env }));
@@ -612,6 +630,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 });
         try { lock?.release(); }
         catch (e) { log(`run ${run}: workspace lock release: ${e.message}`); }
         children.delete(run);

packages/seat/src/compactions.mjs

@@ -0,0 +1,239 @@
+// Row 270 (264c, #1775, T1 item 7 D6): the count of completed compactions over
+// one thread session's lifetime, across its runs and native branches.
+//
+// The host collects it from the session's native store and appends what it
+// found to a ledger, <store dir>/compactions.jsonl (0600, append-only). The
+// store dir holds the mounted session/ and claude-projects/ but is not mounted
+// itself (sandbox.mjs binds()), so the agent can't reach the ledger. The
+// launcher is its only writer, and collects while it holds the session's
+// 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.
+//
+// Ledger lines, each one JSON object with v 1:
+//   { kind: "compaction", entry, completedAt, stream, at }  once per key
+//   { kind: "cursor", stream, bytes, tail, at }  read through `bytes`; `tail`
+//     is the sha256 of the up to 4096 bytes that end there
+//   { kind: "collected", at, known, reason, harness, version }
+// A cursor lets the next collection read only what was appended. A stream
+// 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 { 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;
+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("..");
+
+// The store directory session.mjs sessionStore() makes, as a path only: the
+// reader never creates it. Null for an id that isn't one.
+export function storeDir(dataRoot, business, thread, session, { legacy = false } = {}) {
+  if (![business, thread, session].every(segment)) return null;
+  const dir = join(dataRoot, "threads", business, thread);
+  return legacy ? dir : join(dir, "sessions", session);
+}
+
+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. 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) and store-incomplete (a stream that shrank, changed or
+// went away, or a line that isn't JSON). A missing directory is empty.
+export function scanPi(dir, cursors = {}) {
+  const found = [], next = {}, problems = new Set();
+  let names;
+  try {
+    names = readdirSync(dir);
+  } catch (e) {
+    if (e.code !== "ENOENT") problems.add("store-unreadable");
+    names = [];
+  }
+  for (const stream of Object.keys(cursors)) if (!names.includes(stream)) problems.add("store-incomplete");
+  for (const name of names.filter((n) => n.endsWith(".jsonl")).sort()) {
+    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");
+      }
+      const buf = readRange(fd, from, size);
+      // Complete lines only: a partial trailing line waits for its newline.
+      const end = buf.lastIndexOf(0x0a) + 1;
+      const lines = buf.subarray(0, end).toString("utf8").split("\n");
+      lines.pop();
+      let header = from > 0;
+      for (const line of lines) {
+        if (!line.trim()) continue;
+        let e;
+        try {
+          e = JSON.parse(line);
+        } catch {
+          e = null;
+        }
+        if (!e || typeof e !== "object" || Array.isArray(e)) {
+          problems.add("store-incomplete");
+          continue;
+        }
+        if (!header) {
+          header = true;
+          if (e.type !== "session") problems.add("store-unreadable");
+          continue;
+        }
+        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 });
+        }
+      }
+      const bytes = from + end;
+      next[name] = { bytes, tail: digest(readRange(fd, Math.max(0, bytes - TAIL), bytes)) };
+    } 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) {
+  const events = new Map(), cursors = {};
+  let last = null, incomplete = false, unsupported = false, bad = false, torn = false, text;
+  const file = join(dir, LEDGER);
+  try {
+    if (!lstatSync(file).isFile()) throw Object.assign(new Error(`${file} isn't a regular file`), { code: "ledger-unsafe" });
+    text = readFileSync(file, { encoding: "utf8", flag: fsc.O_RDONLY | fsc.O_NOFOLLOW });
+  } catch (e) {
+    if (e.code === "ENOENT") return null;
+    throw e;
+  }
+  const lines = text.split("\n");
+  if (lines.pop() !== "") bad = torn = true;
+  for (const line of lines) {
+    let r;
+    try {
+      r = JSON.parse(line);
+    } catch {
+      r = null;
+    }
+    if (!r || typeof r !== "object" || r.v !== 1) {
+      bad = true;
+      continue;
+    }
+    if (r.kind === "compaction") {
+      const key = keyOf(r.entry, r.completedAt);
+      if (!events.has(key)) events.set(key, r);
+    } else if (r.kind === "cursor") cursors[r.stream] = { bytes: r.bytes, tail: r.tail };
+    else if (r.kind === "collected") {
+      last = r;
+      if (r.reason === "store-incomplete") incomplete = true;
+      if (r.harness !== "pi") unsupported = true;
+    } else bad = true;
+  }
+  return { events, cursors, last, incomplete, unsupported, bad, torn };
+}
+
+// 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.
+export function compactionCount(dir) {
+  const unknown = (reason, count = null, at = null) => ({ count: count || null, known: false, reason, at });
+  let ledger;
+  try {
+    ledger = readLedger(dir);
+  } catch {
+    return unknown("store-unreadable");
+  }
+  if (!ledger) return unknown("not-collected");
+  const count = ledger.events.size;
+  const at = ledger.last?.at ?? null;
+  if (ledger.bad) return unknown("store-incomplete", count, at);
+  if (!ledger.last) return unknown("not-collected", count);
+  if (ledger.unsupported) return unknown("harness-unsupported", count, at);
+  if (ledger.incomplete) return unknown("store-incomplete", count, at);
+  if (!ledger.last.known) return unknown(ledger.last.reason, count, at);
+  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.
+export function collectCompactions(store, harness, { now = () => new Date().toISOString() } = {}) {
+  const ledger = readLedger(store.dir) ?? { events: new Map(), cursors: {}, torn: false };
+  const at = now();
+  const out = [];
+  if (harness === "pi") {
+    const scan = scanPi(store.session, ledger.cursors);
+    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.
+    const reason = scan.problems.includes("store-incomplete") ? "store-incomplete" : scan.problems[0] ?? 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 });
+  }
+  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" });
+    // After a torn line, start on a fresh one: the torn line stays bad and
+    // keeps the count unknown, and the new lines stay whole.
+    const buf = Buffer.from((ledger.torn ? "\n" : "") + out.map((r) => JSON.stringify(r)).join("\n") + "\n");
+    let off = 0;
+    while (off < buf.length) off += writeSync(fd, buf, off, buf.length - off);
+  } finally {
+    closeSync(fd);
+  }
+  return compactionCount(store.dir);
+}

packages/seat/tests/compactions.test.mjs

@@ -0,0 +1,230 @@
+// 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";
+
+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, 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}`);
+    }
+  }
+});