From 09b20784f6647502b11359e830831e27ec4569b2 Mon Sep 17 00:00:00 2001 From: Jason Woltje Date: Thu, 22 Oct 2026 11:48:19 -0500 Subject: [PATCH] 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 | 113 ++++++++- packages/bus/README.md | 2 +- packages/bus/src/broker.mjs | 21 +- packages/bus/src/runtime.mjs | 6 + .../bus/tests/session-compactions.test.mjs | 87 +++++++ packages/cli/src/launcher.mjs | 19 ++ packages/seat/src/compactions.mjs | 239 ++++++++++++++++++ packages/seat/tests/compactions.test.mjs | 230 +++++++++++++++++ 8 files changed, 710 insertions(+), 7 deletions(-) create mode 100644 packages/bus/tests/session-compactions.test.mjs create mode 100644 packages/seat/src/compactions.mjs create mode 100644 packages/seat/tests/compactions.test.mjs diff --git a/agents/gizmoduck/work/queue-270/BUILD.md b/agents/gizmoduck/work/queue-270/BUILD.md index 2ce11d477..e40f775c7 100644 --- a/agents/gizmoduck/work/queue-270/BUILD.md +++ b/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 - event is the entry `id`. Pi's ids are unique within a file (`generateId(this.byId)`). An entry - id repeated in one file counts once. +- Identity: a compaction is its (entry `id`, entry `timestamp`) across the whole store, not per + stream. Pi's ids are unique only within a file (`generateId(this.byId)`, session-manager.js:21-29), + 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 `/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. diff --git a/packages/bus/README.md b/packages/bus/README.md index 81cb341be..94fa1b300 100644 --- a/packages/bus/README.md +++ b/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:`, 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) diff --git a/packages/bus/src/broker.mjs b/packages/bus/src/broker.mjs index c50f8ac23..c33425486 100644 --- a/packages/bus/src/broker.mjs +++ b/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) - .map((r) => ({ ...r, default: r.id === fallback })); + .all('SELECT id,at,name,predecessor,op FROM sessions WHERE thread=? ORDER BY seq', t.id) + .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 diff --git a/packages/bus/src/runtime.mjs b/packages/bus/src/runtime.mjs index daa0bc65e..6685cfe5f 100644 --- a/packages/bus/src/runtime.mjs +++ b/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, diff --git a/packages/bus/tests/session-compactions.test.mjs b/packages/bus/tests/session-compactions.test.mjs new file mode 100644 index 000000000..bde15c282 --- /dev/null +++ b/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' }); +}); diff --git a/packages/cli/src/launcher.mjs b/packages/cli/src/launcher.mjs index e044b7627..3fefb63bc 100644 --- a/packages/cli/src/launcher.mjs +++ b/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); diff --git a/packages/seat/src/compactions.mjs b/packages/seat/src/compactions.mjs new file mode 100644 index 000000000..ee12592a1 --- /dev/null +++ b/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, /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); +} diff --git a/packages/seat/tests/compactions.test.mjs b/packages/seat/tests/compactions.test.mjs new file mode 100644 index 000000000..5df66084a --- /dev/null +++ b/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}`); + } + } +});