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)
This commit is contained in:
1 parent
8365e4e5b4
commit
09b20784f6
8 files changed
+710
-7
No files matched your search
@@ -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,
|
- 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
|
docs/session-format.md:154), any message inside a compaction's `retainedTail`, and
|
||||||
`branch_summary` entries.
|
`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
|
- Partial trailing line: wait. Only lines that end with a newline are read, as
|
||||||
packages/conversation/src/pi.mjs:35-38 does.
|
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.
|
||||||
@@ -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.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.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. |
|
| `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}`. |
|
| `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)
|
### Projects and threads (row 121, #1615)
|
||||||
|
|||||||
@@ -394,6 +394,7 @@ export class Broker {
|
|||||||
// FED R3: () => the checked access document (R1's loadAccess), read once per user request. Null,
|
// 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).
|
// a throw or no document leaves every user with no grant (design 3.5, fail closed).
|
||||||
#access;
|
#access;
|
||||||
|
#compactions;
|
||||||
// r6 X3: interrupted host grants whose remote undo waits for reconcileHosts, each with its claim.
|
// r6 X3: interrupted host grants whose remote undo waits for reconcileHosts, each with its claim.
|
||||||
#hostUndo = [];
|
#hostUndo = [];
|
||||||
// r6 section 6 (N4): revokes a restart cut short, whose step 3 waits for reconcileHosts.
|
// r6 section 6 (N4): revokes a restart cut short, whose step 3 waits for reconcileHosts.
|
||||||
@@ -419,13 +420,14 @@ export class Broker {
|
|||||||
#clock = 0;
|
#clock = 0;
|
||||||
#processing;
|
#processing;
|
||||||
// credentialStatus is the runtime's metadata-only Credentials.status(); the broker never gets use().
|
// 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');
|
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.
|
// 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 (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 (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 (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 (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.#businesses = structuredClone(businesses);
|
||||||
this.#store = store;
|
this.#store = store;
|
||||||
// A Broker is a trusted host startup, including in-process hosts. Never repair an existing state as off.
|
// 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.#giteaGrant = giteaGrant;
|
||||||
this.#giteaKeys = giteaKeys;
|
this.#giteaKeys = giteaKeys;
|
||||||
this.#access = access;
|
this.#access = access;
|
||||||
|
this.#compactions = compactions;
|
||||||
this.#project = project;
|
this.#project = project;
|
||||||
this.#library = library;
|
this.#library = library;
|
||||||
this.#usage = usage;
|
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');
|
if (s.tester && this.#store.get('SELECT owner FROM threads WHERE id=?', thread).owner !== this.#owner(s)) fail('thread-not-owned');
|
||||||
return thread;
|
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) {
|
#thread(s, id) {
|
||||||
const t = this.#threadRow(s, id);
|
const t = this.#threadRow(s, id);
|
||||||
// REQ-USER-2: a user reaches only their own threads.
|
// 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.
|
// 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).
|
// 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': {
|
case 'session.list': {
|
||||||
if (!s.tester) this.#human(s);
|
if (!s.tester) this.#human(s);
|
||||||
keys(a, ['thread'], ['thread']);
|
keys(a, ['thread'], ['thread']);
|
||||||
const t = this.#thread(s, a.thread);
|
const t = this.#thread(s, a.thread);
|
||||||
const fallback = this.#defaultSession(t.id);
|
const fallback = this.#defaultSession(t.id);
|
||||||
return this.#store
|
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).
|
// 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
|
// The op key makes a retry find the first row: the same op and name answer it, another name
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ import { Credentials } from './credentials.mjs';
|
|||||||
import { serve } from './server.mjs';
|
import { serve } from './server.mjs';
|
||||||
import { launchSettings } from '../../business/src/launch-settings.mjs';
|
import { launchSettings } from '../../business/src/launch-settings.mjs';
|
||||||
import { readSessions } from '../../seat/src/session.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 { publishRemote, workingRepo, WorkingError } from '../../seat/src/working.mjs';
|
||||||
import { createLoader, hasModule } from '../../plugins/src/loader.mjs';
|
import { createLoader, hasModule } from '../../plugins/src/loader.mjs';
|
||||||
import { pluginSettings } from '../../business/src/plugin-settings.mjs';
|
import { pluginSettings } from '../../business/src/plugin-settings.mjs';
|
||||||
@@ -261,6 +262,11 @@ export async function startBroker({
|
|||||||
broker = new Broker({
|
broker = new Broker({
|
||||||
// FED R3: the access file is read per user request; a refusal leaves the user no grant.
|
// 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,
|
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,
|
hostGrant: runRoot ? hostGrant({ configDir: organization.dir, runRoot }) : null,
|
||||||
// Row 235 (#1741, r6 X4, X5): the grant executor's claim, register and issue, never use().
|
// Row 235 (#1741, r6 X4, X5): the grant executor's claim, register and issue, never use().
|
||||||
hostKeys: storeRoot ? grantKeys({ credentials, storeRoot }) : null,
|
hostKeys: storeRoot ? grantKeys({ credentials, storeRoot }) : null,
|
||||||
|
|||||||
@@ -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' });
|
||||||
|
});
|
||||||
@@ -162,6 +162,7 @@ import {
|
|||||||
writeSessionFile,
|
writeSessionFile,
|
||||||
} from "../../seat/src/session.mjs";
|
} from "../../seat/src/session.mjs";
|
||||||
import { BWRAP, sandboxProblem } from "../../seat/src/sandbox.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 { CheckoutError, ensureCheckouts } from "../../seat/src/checkout.mjs";
|
||||||
import { createGitSocket } from "./git-socket.mjs";
|
import { createGitSocket } from "./git-socket.mjs";
|
||||||
import { cmdlineOf, startTimeOf } from "../../seat/src/proc.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) {
|
function serialize(fn) {
|
||||||
const run = chain.then(fn);
|
const run = chain.then(fn);
|
||||||
chain = run.catch(() => {});
|
chain = run.catch(() => {});
|
||||||
@@ -472,6 +489,7 @@ export function createLauncher({
|
|||||||
const { run, runDir } = newRun(dataRoot, b.id);
|
const { run, runDir } = newRun(dataRoot, b.id);
|
||||||
const sessionDir = store ? store.session : join(runDir, "session");
|
const sessionDir = store ? store.session : join(runDir, "session");
|
||||||
if (!store) mkdirSync(sessionDir, { mode: 0o700 });
|
if (!store) mkdirSync(sessionDir, { mode: 0o700 });
|
||||||
|
if (store) compactions(store, model.harness, { instance, thread, session, run });
|
||||||
const toolSocket = join(runDir, "tools.sock");
|
const toolSocket = join(runDir, "tools.sock");
|
||||||
const turnMarker = join(runDir, "turn.done");
|
const turnMarker = join(runDir, "turn.done");
|
||||||
const version = harnessVersion !== undefined ? harnessVersion : call(model.harness, "version", () => harness.version({ env }));
|
const version = harnessVersion !== undefined ? harnessVersion : call(model.harness, "version", () => harness.version({ env }));
|
||||||
@@ -612,6 +630,7 @@ export function createLauncher({
|
|||||||
} finally {
|
} finally {
|
||||||
try { unregisterRuntime(dataRoot, managed.id); }
|
try { unregisterRuntime(dataRoot, managed.id); }
|
||||||
catch { log(`run ${run}: processing receipt cleanup refused`); }
|
catch { log(`run ${run}: processing receipt cleanup refused`); }
|
||||||
|
if (store) compactions(store, model.harness, { instance, thread, session, run });
|
||||||
try { lock?.release(); }
|
try { lock?.release(); }
|
||||||
catch (e) { log(`run ${run}: workspace lock release: ${e.message}`); }
|
catch (e) { log(`run ${run}: workspace lock release: ${e.message}`); }
|
||||||
children.delete(run);
|
children.delete(run);
|
||||||
|
|||||||
@@ -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);
|
||||||
|
}
|
||||||
@@ -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}`);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
Reference in new issue
Block a user