gizmoduck: row 270 (264c, #1775) round 2 for duckworth-0223 B1-B3 (sage-3030, sage-3031, mos-1352, sage-3033): pi streams read in 1 MiB chunks under a budget from the system config's compactionReadBytes (default 64 MiB, 1 MiB to 1 GiB, a bad value takes the default and logs), not-collected past it; Claude history from before the ledger is harness-unsupported for good; launcher, runtime reader, clear and legacy tests; mutants M13-M21; BUILD.md round 2 with sizes, schema change, rollback hazard, N1-N3; not run

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

@@ -194,3 +194,170 @@ T3 Pi half as built; the Tests section above never claimed the wiring. Adding a
 test is a scope call for Sage.
 
 Prior coverage: none. 271's M1-M14 edit no 270 line, and no past verdict names 270.
+
+## Round 2 (duckworth-0223 B1-B3, sage-3030, sage-3031, mos-1352)
+
+New commits on 8a5eb3c8b (round 1, on 7d199f033). Nothing here has run: no test, no mutant, no node.
+Each kill below is read from the test.
+
+### B1: the read budget, a human-only setting
+
+`scanPi` reads a stream a CHUNK (1 MiB) at a time from its cursor, carries a line split across
+reads, and stops when the collection's budget runs out, over all streams together. The whole-range
+`Buffer.alloc(to - from)` that compactions.mjs:57 and :101 had is gone from the stream read;
+`readRange` now serves only the tail digest reads, at most 4096 bytes. A collection the budget stops
+short records `known: false`, reason `not-collected`, and cursors at the last whole line it read. The
+reason isn't sticky: `compactionCount` takes `last.reason`, so a later collection that reads to the
+end reads known again. When a collection has more than one problem the order is store-incomplete,
+store-unreadable, then not-collected.
+
+The setting (mos-1352, DECISIONS Q222). It's `compactionReadBytes`, a top-level key of the instance's
+system config, the file `MOSAIC_CONFIG` names (`~/.config/mosaic-NAME/config.json`). Under invariant 2
+a human writes that file and nothing auto-writes it. The launcher reads it once, at host start,
+through `loadSystem`; nothing reads it from the session mount, the workspace or an environment
+variable an agent's process sets. `readBudget` (compactions.mjs) takes a whole number of bytes from
+1 MiB to 1 GiB and gives the default for anything else, missing included. The launcher logs a bad
+value once at start, naming the default. A change takes effect at the next host start.
+
+The schema change, for review. `scripts/mosaic-config.mjs` refused any top-level key outside
+`configVersion, environment, dataRoot, execution, console` (`unsupported configuration key`). It now
+also takes `compactionReadBytes`, and passes it on as written, a bad value included, because the
+fallback is the launcher's and a bad value must not refuse the config. The output carries the key
+only when the file has it, so a config without it prints what it always did (test-config.sh's
+"config without console" case still pins that). The old list literal stays as it was and the key is
+added by `.concat`, because packages/cli/tests/stack-release-fixture.mjs:300 edits that literal to
+make an older loader.
+
+A consequence for Duckworth to rule on. A loader from before this round refuses a config that sets
+the key. scripts/stack-release.sh `config_checks` (:1079-1097) runs the target copy's loader only when
+the config has a `console` key. So a human who sets `compactionReadBytes` and then rolls back to a
+copy from before 270 passes that check, and the bus fails at its next config read. Nobody has to set
+the key, since the default needs none. I've left stack-release.sh alone: it's outside this row's
+manifest. The bounded follow-up is to make `config_checks` run the target loader whenever the config
+has any key the target might refuse, not just `console`.
+
+The default, 64 MiB, from sizes measured 2026-10-22 by stat and line length only (no content printed,
+nothing under the live stack read):
+
+| Store | Files | p50 | p90 | p99 | max |
+| --- | --- | --- | --- | --- | --- |
+| Pi sessions, `~/.pi/agent/sessions` | 224 | 298 KB | 1.1 MB | 4.7 MB | 11.6 MB |
+| Claude Code transcripts, `~/.claude/projects` (not read by the scanner) | 549 | 607 KB | 6.6 MB | 120 MB | 603 MB |
+
+The longest Pi line is 0.84 MB, and no Pi line is over 1 MiB. 64 MiB is more than five times the
+largest Pi stream, so a session's first collection reads it whole, and later ones read only what
+was appended. Memory per collection is one 1 MiB chunk plus the carry of one unfinished line, and
+the budget bounds both. The Claude numbers show how far a long agent session can grow. A Pi session
+that size would take several collections to read through, each one reading unknown until the last.
+I haven't measured how long a 64 MiB collection takes; the sparse test reads 64 MiB of a hole, and its
+wall time in the slot run will be the first number.
+
+Known limits:
+- A single line longer than the budget is never read past. The count stays not-collected until a
+  human raises the setting. The longest Pi line measured is 0.84 MB.
+- `readLedger` still reads the ledger whole; B1 named the stream reads. Ledger lines are 120 bytes
+  (collected), 226 (cursor) and 199 (compaction), so a run with two collections, one changed stream
+  and one compaction adds about 0.9 KB. One session reaches 64 MiB of ledger after about 75,000 runs.
+
+### B3: Claude history from before the ledger
+
+When a Pi collection finds no ledger and the store's `claude-projects/` (`store.projects`) isn't
+empty, it skips the scan and writes one collected record: `known: false`, reason
+`harness-unsupported`, harness `pi`. `readLedger` now sets `unsupported` for a record whose harness
+isn't `pi` or whose reason is `harness-unsupported`, so the count stays unknown for good, as after a
+Claude run. A `claude-projects/` that can't be read throws, so nothing is recorded and the next
+collection checks again. An empty one is no history. Once a ledger exists the check doesn't run: a
+later Claude run collects for itself.
+
+A Pi run mounts `claude-projects/` read-write (sandbox.mjs:125). The run-start collection comes before
+the agent's first turn, though, so an agent can't put a file there before a fresh session's first
+collection. The exception is a run whose start collection failed. Even then, the agent can only
+make the count unknown, never wrong.
+
+The Vikunja 466 reason. For a Claude run, seat compactions.test.mjs:199 asserts `harness-unsupported`,
+and :201 asserts that it stays after a Pi run. For history from before the ledger,
+compactions-budget.test.mjs:88 and :89 assert it (sage-3030 item 2).
+
+### B2: the run wiring, the reader and a clear
+
+The new tests live in files 271 doesn't touch:
+- `packages/seat/tests/compactions-budget.test.mjs`, B1 and B3.
+- `packages/cli/tests/compactions-launch.test.mjs`, with its stub adapter
+  `packages/cli/tests/fixtures/compaction-adapter.mjs`, real runners in bwrap.
+
+compactions-launch tests:
+1. A run's start and end collections, across a resume and two host restarts. session.list reads the
+   ledger through the runtime's reader (:166-198).
+2. A collection that fails, with the ledger's path made a directory. Each failure is logged as
+   `run X: compactions: ...` and recorded as `store-unreadable`. The run answers, and it ends with the
+   same reason, exit code and signal as the next run, whose collections count (:200-224).
+3. The setting. At 1 MiB, a 2 MiB hole is not-collected at both collections. With `"x"`, the host
+   logs the fallback once, and the default reads the hole through (:226-261).
+4. A clear. The retired session keeps its count, 1, and the successor reads `{count: 0, known: true}`
+   after its first collection. A legacy session (op legacy-3l, seeded with the SQL from
+   bus session-compactions.test.mjs:72-74) reads the thread store's ledger through the runtime's reader
+   after a restart (:263-308).
+
+### Mutants M13-M21
+
+`mutants.py`, with all 21 anchors matching once at this round's tip (`check`, python only).
+
+| Mutant | Edit | Kill: first failing assertion |
+| --- | --- | --- |
+| M13 | drop the run-start `compactions(...)` call (launcher.mjs:501) | cli compactions-launch.test.mjs:182, run 1 logs only `[D,2,true,null]` |
+| M14 | drop the run-end call (launcher.mjs:642) | :182, run 1 logs only `[D,1,true,null]` |
+| M15 | the runtime's reader returns `null` (runtime.mjs:268) | :197, compactions null; :286 and :306 fail the same way |
+| M16 | no budget, `stop = size` | seat compactions-budget.test.mjs:48, the hole is read whole: known true |
+| M17 | `readBudget` returns `value ?? READ_BYTES` | :37, `"8388608"` comes back as the string |
+| M18 | the launcher ignores the setting, `budget = READ_BYTES` | cli :248, the 2 MiB hole reads known twice |
+| M19 | no carry, `take(view.subarray(i, nl))` | seat :79, c1's halves fail to parse: store-incomplete |
+| M20 | no Claude history check | :88, c1 counts known |
+| M21 | `harness-unsupported` not sticky on a `pi` record | :89, the second collection reads known |
+
+The files they touch, sha256 at this round's tip, to check each restore against (sage-3030 item 3):
+
+```
+45a3744598cfb8aa1cfcffd8a0ee2dd2bb631061efd192d8461f1901141993ee  packages/seat/src/compactions.mjs
+d7cdc8a17e337e00dfe2d1bf6249cc2c2e696b16c5b426b2fe616089e5559509  packages/bus/src/broker.mjs
+5d36cc0fa19ab8e478ecf9a807e3afab8181cb650778e80b13efe785e6c44995  packages/cli/src/launcher.mjs
+5ae423438a178b1c8eac3b8ce0e0c22f5bb7caebb9011ce7098a36d95b53f1ca  packages/bus/src/runtime.mjs
+```
+
+broker.mjs and runtime.mjs are byte-identical to 7d199f033.
+
+### Run plan, on a CLEAR only
+
+Every run uses TMPDIR from `mktemp -d /mnt/storage/scratch/tmp/gXXX` and `timeout -k 10`. A first-run
+red in a new test is placed and reported after the run (mos-1272). A red in an old test, a product
+fix or a second red goes to Sage and Mos first.
+
+1. Each mutant runs its kill file whole: M1-M10 seat compactions.test.mjs, M16, M17 and M19-M21 seat
+   compactions-budget.test.mjs (120 s each), M11-M12 bus session-compactions.test.mjs (120 s), and
+   M13-M15 and M18 cli compactions-launch.test.mjs (600 s). After each one: `git checkout` the file,
+   check its sha256 against the table, and confirm the tree is clean. A byte-identical restore needs
+   no other rerun.
+2. The new test files, whole: seat compactions-budget.test.mjs and cli compactions-launch.test.mjs.
+3. Every file that reads a path this round touches, whole:
+   - compactions.mjs: seat compactions.test.mjs, bus session-compactions.test.mjs.
+   - mosaic-config.mjs: scripts/test-config.sh, cli stack-instances.test.mjs, cli stack-release.test.mjs
+     (through stack-release-fixture.mjs), webui pinned-copy.test.mjs, and the business suite (it
+     writes configs through helpers.mjs:79).
+   - launcher.mjs: the cli files that import it, all in the cli suite.
+   - README.md: no test reads it.
+4. The gate (sage-3031) runs on the combined 270+271 tip in one slot after Gander's 276 run5. It
+   covers bus, seat and cli whole, the shell suites, M1-M21 and 271's mutants. compactions.mjs is in
+   271's manifest, so 271's earlier gate evidence is stale. Webui pinned-copy and the business suite
+   join it because the loader changed.
+
+### Known residuals (duckworth-0223, sage-3031)
+
+- N1. A session.list read can briefly see a half-written line. The bus reads without the workspace
+  lock, so a read during an append sees a torn last line. That call alone reports store-incomplete,
+  and nothing is persisted. Accepted.
+- N2 (README). A launcher killed mid-write leaves a torn ledger line, and the count stays unknown
+  (store-incomplete) for good, by design (sage-2806 condition 3). This is now a line in the bus
+  README's `session.list` row.
+- N3. Pi 0.85.1 rewrites a session file only when it's empty (session-manager.js:625-632), on a
+  version migration (:675-677), or for a new branched file (:1173). None of these moves bytes under a
+  0.85.1 cursor, so the 4096-byte tail check is enough. A non-Pi writer that rewrites earlier bytes and
+  keeps the tail goes unseen. That residual is acceptable.

agents/gizmoduck/work/queue-270/mutants.py

@@ -1,13 +1,20 @@
 #!/usr/bin/env python3
-# Row 270 mutants M1-M12 (BUILD.md "Mutants", anchors at 7d199f033). `apply <id>` edits the
-# worktree and prints the touched path; each edit asserts its anchor appears exactly once.
-# K names the test and the first assertion each mutant should fail (checked by reading, not run).
+# Row 270 mutants. M1-M12 are round 1's (anchors at 7d199f033); M13-M21 are round 2's
+# (duckworth-0223 B1-B3, sage-3030, sage-3031, mos-1352). `apply <id>` edits the worktree and
+# prints the touched path; each edit asserts its anchor appears exactly once. `check` prints each
+# anchor's count and kill. K names the test and the first assertion each mutant should fail
+# (checked by reading, not run). The base blob of every touched file is in BUILD.md; a restore is
+# `git checkout -- <path>`, then sha256 against it (sage-3030 item 3).
 import sys
 WT = "/mnt/storage/scratch/gizmo-270/"
 C = "packages/seat/src/compactions.mjs"
 B = "packages/bus/src/broker.mjs"
+L = "packages/cli/src/launcher.mjs"
+R = "packages/bus/src/runtime.mjs"
 ST = "packages/seat/tests/compactions.test.mjs"
+SB = "packages/seat/tests/compactions-budget.test.mjs"
 BT = "packages/bus/tests/session-compactions.test.mjs"
+CT = "packages/cli/tests/compactions-launch.test.mjs"
 M = {
  "M1": (C, "      const key = keyOf(f.entry, f.completedAt);\n      if (ledger.events.has(key)) continue;",
            "      const key = keyOf(f.stream + f.entry, f.completedAt);\n      if (ledger.events.has(key)) continue;"),
@@ -25,6 +32,24 @@ M = {
  "M11": (B, "op === 'legacy-3l')", "false)"),
  "M12": (B, "if (!this.#compactions) return unknown('not-collected');",
             "if (!this.#compactions) return { count: 0, known: true, reason: null, at: null };"),
+ # B2: the launcher's two collections and the runtime's reader.
+ "M13": (L, "    if (store) compactions(store, model.harness, { instance, thread, session, run });\n    const toolSocket",
+            "    const toolSocket"),
+ "M14": (L, "        if (store) compactions(store, model.harness, { instance, thread, session, run });\n        try { lock?.release(); }",
+            "        try { lock?.release(); }"),
+ "M15": (R, "        return dir ? compactionCount(dir) : { count: null, known: false, reason: 'store-unreadable', at: null };",
+            "        return null;"),
+ # B1: the budget, its setting and the carry across reads.
+ "M16": (C, "      const stop = Math.min(size, from + left);", "      const stop = size;"),
+ "M17": (C, "  return Number.isSafeInteger(value) && value >= READ_MIN && value <= READ_MAX ? value : READ_BYTES;",
+            "  return value ?? READ_BYTES;"),
+ "M18": (L, "  const budget = readBudget(system?.compactionReadBytes);", "  const budget = READ_BYTES;"),
+ "M19": (C, "          take(carry.length ? Buffer.concat([...carry, view.subarray(i, nl)]) : view.subarray(i, nl));",
+            "          take(view.subarray(i, nl));"),
+ # B3: Claude history from before the ledger, and its stickiness.
+ "M20": (C, '  if (harness === "pi" && !prior && claudeHistory(store)) {', '  if (harness === "pi" && !prior && false) {'),
+ "M21": (C, 'if (r.harness !== "pi" || r.reason === "harness-unsupported") unsupported = true;',
+            'if (r.harness !== "pi") unsupported = true;'),
 }
 K = {
  "M1": (ST, "a compaction copied into a branched stream counts once, under the stream it was first seen in", 64,
@@ -48,6 +73,24 @@ K = {
  "M11": (BT, "a legacy session is asked for with legacy true", 76, "asked with legacy false"),
  "M12": (BT, "without a reader the count is unknown, never 0; a reader that throws is store-unreadable", 81,
          "{count 0, known true} for no reader"),
+ "M13": (CT, "a run collects compactions at its start and its end, and the count holds across a resume and a host restart", 182,
+         "no start collection: run 1 logs only [D,2,true,null]; settle returns after 5 s with one event"),
+ "M14": (CT, "a run collects compactions at its start and its end, and the count holds across a resume and a host restart", 182,
+         "no end collection: run 1 logs only [D,1,true,null]"),
+ "M15": (CT, "a run collects compactions at its start and its end, and the count holds across a resume and a host restart", 197,
+         "session.list's compactions is null: {at: 'undefined'} and no count; the clear test's :286 and the legacy :306 fail the same way"),
+ "M16": (SB, "a stream past the default budget is not-collected with a lower bound, and a later collection catches up", 48,
+         "the whole 66 MiB is read: {count 1, known true, reason null}"),
+ "M17": (SB, "the budget is the setting's whole number of bytes from 1 MiB to 1 GiB, and the default for anything else", 37,
+         "readBudget(null) is the default, but readBudget('8388608') is the string"),
+ "M18": (CT, "the read budget is the system config's compactionReadBytes, and a bad value takes the default", 248,
+         "the 1 MiB setting is ignored, the 2 MiB hole is read whole: [D,1,true,null] twice"),
+ "M19": (SB, "a line split across reads counts once", 79,
+         "c1's two halves parse alone and fail: {count null, known false, reason store-incomplete}"),
+ "M20": (SB, "Claude history from before the ledger makes the first Pi collection harness-unsupported, for good", 88,
+         "the first collection counts c1: {count 1, known true}"),
+ "M21": (SB, "Claude history from before the ledger makes the first Pi collection harness-unsupported, for good", 89,
+         "a Pi collected record isn't sticky: the second reads {count 1, known true}"),
 }
 cmd = sys.argv[1]
 if cmd == "list":

packages/bus/README.md

@@ -246,7 +246,7 @@ a receipt with empty `changes`. A refusal is `action.refused` with its code alon
 | `thread.messages` | `{thread,session?,after?,limit?}`; human sessions only. The session's messages oldest first, after the message `after` names (`message-not-found` when it isn't in the session), at most `limit` (default 50, 1 to 200). A thread shared with the caller reads too, any session of it (row 206). Writes nothing. |
 | `thread.share`, `thread.unshare` | `{thread,to}`; the thread's owner only (row 206, #1704). `to` is `user:<id>`, the business human's name or `*`; yourself refuses `invalid-request`. A share refuses `share-exists` when one is in force, and `share-not-covered` when a `user:` target has no live grant covering the thread. An unshare refuses `share-not-found` when none is. Each writes one `thread_shares` row. Returns `{thread,to,op}`. |
 | `thread.shares` | `{thread}`; the thread's owner only. The shares in force, `[{to,at,by}]`, oldest first. Writes nothing. |
-| `session.list` | `{thread}`; human and tester sessions (row 264a, #1769). The thread's sessions oldest first, `[{id,at,name,predecessor,default,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.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`. `not-collected` passes: the next collection reads on. A launcher killed mid-write leaves a torn ledger line, and the count stays unknown (`store-incomplete`) for good, by design (sage-2806 condition 3). Writes nothing. |
 | `session.create` | `{thread,name,op}`; human and tester sessions. A new session of the thread, metadata only. A retry with the same `op` and name returns the first row; another name under that `op` refuses `session-op-conflict`, and a taken name `session-exists`. Returns `{id,thread,name}`. |
 
 ### Projects and threads (row 121, #1615)

packages/cli/src/launcher.mjs

@@ -162,7 +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 { collectCompactions, READ_BYTES, readBudget } 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";
@@ -291,11 +291,17 @@ export function createLauncher({
   // 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.
+  // reads unknown with its reason. Row 270 r2 (B1, mos-1352): each collection
+  // reads at most the system config's compactionReadBytes, a human-only
+  // setting; a missing or bad value takes the default.
+  const budget = readBudget(system?.compactionReadBytes);
+  if (system?.compactionReadBytes !== undefined && budget !== system.compactionReadBytes) {
+    log(`system config: compactionReadBytes ${JSON.stringify(system.compactionReadBytes)} isn't a whole number of bytes from 1048576 to 1073741824; compaction collections read up to the default ${READ_BYTES}`);
+  }
   const compactions = (store, harness, { instance, thread, session, run }) => {
     let c;
     try {
-      c = collectCompactions(store, harness);
+      c = collectCompactions(store, harness, { budget });
     } catch (e) {
       log(`run ${run}: compactions: ${e.message}`);
       c = { count: null, known: false, reason: "store-unreadable" };

packages/cli/tests/compactions-launch.test.mjs

@@ -0,0 +1,308 @@
+// Compaction counts on a real run (row 270, 264c, #1775, T3's Pi half): the
+// launcher collects a thread session's completed compactions into its
+// store's ledger at run start and at run end, and session.list reads the
+// ledger through the runtime's reader. The count holds across a resume and a
+// host restart. Real runners in the bwrap sandbox; a stub Pi adapter writes
+// the session stream (fixtures/compaction-adapter.mjs).
+//
+// Round 2 (duckworth-0223 B2, sage-3031, mos-1352): a failed collection, the
+// read budget from the system config, a clear's retired session and its
+// successor, and a legacy session read through the runtime's reader.
+//
+// The host is session-launch.test.mjs's: startBroker on bootConfig's config,
+// with the ops the broker child forwards, and the human capability from
+// bindHuman directly.
+
+import { test } from "node:test";
+import assert from "node:assert/strict";
+import { spawnSync } from "node:child_process";
+import { existsSync, mkdirSync, readFileSync, rmSync, truncateSync, writeFileSync } from "node:fs";
+import { join } from "node:path";
+import { fileURLToPath } from "node:url";
+import { launchDoc } from "../../business/tests/helpers.mjs";
+import { BusError } from "../../bus/src/broker.mjs";
+import { startBroker } from "../../bus/src/runtime.mjs";
+import { Store } from "../../bus/src/store.mjs";
+import { collectCompactions, LEDGER } from "../../seat/src/compactions.mjs";
+import { gitEnv } from "../../seat/src/checkout.mjs";
+import { sandboxProblem } from "../../seat/src/sandbox.mjs";
+import { launchLogFile, SESSION_CREDENTIALS, sessionStore } from "../../seat/src/session.mjs";
+import { bootConfig, loadSystem } from "../src/config.mjs";
+import { createLauncher } from "../src/launcher.mjs";
+import { fakeHarnesses, fixture, tmp } from "./helpers.mjs";
+
+const ADAPTER = fileURLToPath(new URL("./fixtures/compaction-adapter.mjs", import.meta.url));
+const skip = sandboxProblem() ?? false;
+
+async function until(fn, ms = 20000) {
+  const end = Date.now() + ms;
+  while (Date.now() < end) {
+    const v = await fn();
+    if (v) return v;
+    await new Promise((r) => setTimeout(r, 50));
+  }
+  throw new Error("timed out waiting");
+}
+// A mutant's kill is an assertion failure, not a timeout (sage-2707), so a wait a mutant could
+// stall returns what it last saw and the assertions after it decide.
+async function settle(fn, ms = 5000) {
+  const end = Date.now() + ms;
+  let v;
+  while (Date.now() < end) {
+    v = await fn();
+    if (v) return v;
+    await new Promise((r) => setTimeout(r, 50));
+  }
+  return v;
+}
+
+function gitIn(cwd, ...args) {
+  const r = spawnSync("git", ["-c", "user.name=test", "-c", "[email protected]", "-C", cwd, ...args], { env: gitEnv(), encoding: "utf8" });
+  assert.equal(r.status, 0, r.stderr);
+}
+
+// The assistant is the thread instance, on Pi.
+function prepare(t) {
+  const root = tmp(t, "mosaic-compactions-");
+  const f = fixture(root, "acme", (doc) => {
+    doc.roles.chief.vars = { harness: "pi", model: "claude-opus-5-5", thinking: "high" };
+    doc.roles.assistant = { definition: "assistant", vars: { harness: "pi", model: "claude-sonnet-5-5", thinking: "high" } };
+    doc.launch = launchDoc("chief", ["coder", "assistant"], { opus: 1, sonnet: 3 });
+    doc.launch.threadIdleSeconds = 1;
+    return doc;
+  });
+  for (const k of SESSION_CREDENTIALS.flat()) delete f.env[k];
+  const project = f.doc.projects.stack.root;
+  gitIn(project, "init", "-q", "-b", "main");
+  writeFileSync(join(project, "file.txt"), "one\n");
+  gitIn(project, "add", "file.txt");
+  gitIn(project, "commit", "-q", "-m", "one");
+  const adapter = join(root, "adapter.sh");
+  writeFileSync(adapter, `exec '${process.execPath}' '${ADAPTER}'\n`, { mode: 0o644 });
+  return { root, f, adapter };
+}
+
+// A host and its launcher on the prepared root.
+async function start(t, { f, adapter }) {
+  const system = loadSystem({ env: f.env });
+  const { trackers, ...boot } = bootConfig({ system, businessId: "acme", env: f.env });
+  assert.equal(trackers, undefined);
+  const runtime = await startBroker(boot);
+  const ops = {
+    identity: (m) => runtime.identity(m.cap),
+    authorizeLaunch: (m) => runtime.authorizeLaunch(m.cap, m.instance),
+    refuse: (m) => runtime.refuse(m.cap, m.code, m.subject),
+    endLaunch: (m) => runtime.endLaunch(m.record),
+    credentialStatus: (m) => runtime.credentialStatus(m.business, m.role),
+    threadsState: (m) => runtime.threadsState(m.business),
+    threadIdle: (m) => runtime.threadIdle(m.record),
+    mayLaunch: (m) => runtime.mayLaunch(m),
+    landHeld: (m) => runtime.landHeld(m.record),
+    landHolders: (m) => runtime.landHolders(m.business),
+    landRelease: (m) => runtime.landRelease(m.record),
+  };
+  const refusing = (what, fn) => async (m) => {
+    try {
+      return fn(m);
+    } catch (e) {
+      const code = e instanceof BusError ? e.code : `${what}-refused`;
+      throw Object.assign(new Error(`${what} refused: ${code}`), { code });
+    }
+  };
+  const hooks = [];
+  let closing = null;
+  const host = {
+    business: "acme",
+    path: runtime.path,
+    dataRoot: f.dataRoot,
+    op: refusing("launch-op", (m) => {
+      if (!Object.hasOwn(ops, m.op)) throw new BusError("invalid-host-request");
+      return ops[m.op](m);
+    }),
+    bindLaunch: refusing("launch bind", (r) => runtime.bindLaunch(r)),
+    beforeClose: (fn) => void hooks.push(fn),
+  };
+  const logs = [];
+  const launcher = createLauncher({
+    host,
+    system,
+    env: f.env,
+    makeLoader: await fakeHarnesses(t, { pi: adapter, "claude-code": adapter }),
+    sessionDefaults: { pollInterval: 100 },
+    harnessVersion: "0.85.1",
+    readOnly: [ADAPTER],
+    threadTickMs: 100,
+    log: (l) => logs.push(l),
+  });
+  const close = () =>
+    (closing ??= (async () => {
+      for (const fn of hooks) await fn(0);
+      await runtime.close();
+    })());
+  t.after(close);
+  await launcher.listen();
+  const human = runtime.broker.bindHuman({ business: "acme", human: f.doc.human, via: "cli", outsideAgent: true });
+  const call = (verb, args = {}) => runtime.broker.request(human, { verb, args });
+  return { runtime, launcher, call, logs, close };
+}
+
+const launchLog = (f) =>
+  existsSync(launchLogFile(f.dataRoot, "acme"))
+    ? readFileSync(launchLogFile(f.dataRoot, "acme"), "utf8").trim().split("\n").map((l) => JSON.parse(l))
+    : [];
+const events = (f, event) => launchLog(f).filter((e) => e.event === event);
+const counted = (f) => events(f, "compactions").map((e) => [e.session, e.count, e.known, e.reason]);
+
+// Send to a thread's default session and wait for the assistant's RESULT reply.
+async function ask(call, thread, body) {
+  const sent = call("thread.send", { thread, body });
+  const r = await until(() => call("thread.messages", { thread }).find((m) => m.in_reply_to === sent.id && m.class === "RESULT"));
+  assert.equal(r.from_role, "assistant");
+  return JSON.parse(r.body.match(/THREAD (\{.*\})/)[1]);
+}
+
+const line = (e) => `${JSON.stringify(e)}\n`;
+
+test("REQ-THREAD-2 T3 a run collects compactions at its start and its end, and the count holds across a resume and a host restart", { skip, timeout: 90000 }, async (t) => {
+  const prepared = prepare(t);
+  const { f } = prepared;
+  let host = await start(t, prepared);
+  const project = host.call("project.create", { name: "stack work", repo: "stack" }).id;
+  const A = host.call("thread.create", { project, name: "alpha" }).id;
+  const [{ id: D }] = host.call("session.list", { thread: A });
+  // An earlier run left a compaction no run-end collection saw (a host that died mid-run).
+  // Only the run-start collection counts it before the agent's first turn.
+  const store = sessionStore(f.dataRoot, "acme", A, D);
+  writeFileSync(join(store.session, "pi.jsonl"), line({ type: "session", version: 3, id: "pi", timestamp: "2026-10-22T11:00:00.000Z", cwd: "/w" }) + line({ type: "compaction", id: "c0", parentId: null, timestamp: "2026-10-22T10:59:00.000Z", summary: "s", firstKeptEntryId: "k", tokensBefore: 100 }), { mode: 0o600 });
+
+  // Run 1: start counts c0 (M13), the turn compacts c1, the end counts it (M14).
+  assert.deepEqual(await ask(host.call, A, "compact c1"), { body: "compact c1", before: [] });
+  await until(() => events(f, "end").length === 1, 30000);
+  await settle(() => counted(f).length >= 2);
+  assert.deepEqual(counted(f), [[D, 1, true, null], [D, 2, true, null]]);
+
+  // A restart: a new host and launcher on the same data root. Run 2 resumes the session.
+  await host.close();
+  host = await start(t, prepared);
+  assert.deepEqual(await ask(host.call, A, "compact c2"), { body: "compact c2", before: ["compact c1"] });
+  await until(() => events(f, "end").length === 2, 30000);
+  await settle(() => counted(f).length >= 4);
+  assert.deepEqual(counted(f), [[D, 1, true, null], [D, 2, true, null], [D, 2, true, null], [D, 3, true, null]]);
+
+  // session.list reads the ledger through the runtime's reader (M15), after another restart.
+  await host.close();
+  host = await start(t, prepared);
+  const [row] = host.call("session.list", { thread: A });
+  assert.equal(row.id, D);
+  assert.deepEqual({ ...row.compactions, at: typeof row.compactions?.at }, { count: 3, known: true, reason: null, at: "string" });
+});
+
+test("REQ-THREAD-2 T3 a collection that fails is logged and recorded store-unreadable, and the run ends as one whose collections count", { skip, timeout: 90000 }, async (t) => {
+  const prepared = prepare(t);
+  const { f } = prepared;
+  const host = await start(t, prepared);
+  const project = host.call("project.create", { name: "stack work", repo: "stack" }).id;
+  const A = host.call("thread.create", { project, name: "alpha" }).id;
+  const [{ id: D }] = host.call("session.list", { thread: A });
+  // The ledger's path is a directory, so readLedger refuses it (ledger-unsafe).
+  const store = sessionStore(f.dataRoot, "acme", A, D);
+  mkdirSync(join(store.dir, LEDGER));
+  assert.deepEqual(await ask(host.call, A, "compact c1"), { body: "compact c1", before: [] });
+  await until(() => events(f, "end").length === 1, 30000);
+  await settle(() => counted(f).length >= 2);
+  assert.deepEqual(counted(f), [[D, null, false, "store-unreadable"], [D, null, false, "store-unreadable"]]);
+  const [{ run: first }] = events(f, "launch");
+  assert.equal(host.logs.filter((l) => l.startsWith(`run ${first}: compactions: `)).length, 2);
+  // With the ledger's path free again, the next run's collections count what the stream holds.
+  rmSync(join(store.dir, LEDGER), { recursive: true });
+  assert.deepEqual(await ask(host.call, A, "compact c2"), { body: "compact c2", before: ["compact c1"] });
+  await until(() => events(f, "end").length === 2, 30000);
+  await settle(() => counted(f).length >= 4);
+  assert.deepEqual(counted(f).slice(2), [[D, 1, true, null], [D, 2, true, null]]);
+  const ends = events(f, "end").map((e) => [e.reason, e.exitCode, e.signal]);
+  assert.deepEqual(ends[0], ends[1]);
+});
+
+test("REQ-THREAD-2 T3 the read budget is the system config's compactionReadBytes, and a bad value takes the default", { skip, timeout: 90000 }, async (t) => {
+  const prepared = prepare(t);
+  const { f } = prepared;
+  const setting = (v) => {
+    const doc = JSON.parse(readFileSync(f.config, "utf8"));
+    doc.compactionReadBytes = v;
+    writeFileSync(f.config, JSON.stringify(doc));
+  };
+  setting(1 << 20);
+  let host = await start(t, prepared);
+  const project = host.call("project.create", { name: "stack work", repo: "stack" }).id;
+  const A = host.call("thread.create", { project, name: "alpha" }).id;
+  const [{ id: D }] = host.call("session.list", { thread: A });
+  // c0, then a 2 MiB hole with no newline: a line past 1 MiB, inside the 64 MiB default.
+  const store = sessionStore(f.dataRoot, "acme", A, D);
+  const path = join(store.session, "pi.jsonl");
+  const whole = line({ type: "session", version: 3, id: "pi", timestamp: "2026-10-22T11:00:00.000Z", cwd: "/w" }) + line({ type: "compaction", id: "c0", parentId: null, timestamp: "2026-10-22T10:59:00.000Z", summary: "s", firstKeptEntryId: "k", tokensBefore: 100 });
+  writeFileSync(path, whole, { mode: 0o600 });
+  truncateSync(path, Buffer.byteLength(whole) + 2 * (1 << 20));
+  assert.deepEqual(await ask(host.call, A, "one"), { body: "one", before: [] });
+  await until(() => events(f, "end").length === 1, 30000);
+  await settle(() => counted(f).length >= 2);
+  assert.deepEqual(counted(f), [[D, 1, false, "not-collected"], [D, 1, false, "not-collected"]]);
+  assert.deepEqual(host.logs.filter((l) => l.startsWith("system config: ")), []);
+
+  setting("x");
+  await host.close();
+  host = await start(t, prepared);
+  assert.deepEqual(host.logs.filter((l) => l.startsWith("system config: ")), [
+    'system config: compactionReadBytes "x" isn\'t a whole number of bytes from 1048576 to 1073741824; compaction collections read up to the default 67108864',
+  ]);
+  assert.deepEqual(await ask(host.call, A, "two"), { body: "two", before: ["one"] });
+  await until(() => events(f, "end").length === 2, 30000);
+  await settle(() => counted(f).length >= 4);
+  assert.deepEqual(counted(f).slice(2), [[D, 1, true, null], [D, 1, true, null]]);
+});
+
+test("REQ-THREAD-2 T3 T4 a clear leaves the retired session's count as it was, its successor counts from 0, and a legacy session reads the thread's ledger", { skip, timeout: 90000 }, async (t) => {
+  const prepared = prepare(t);
+  const { f } = prepared;
+  let host = await start(t, prepared);
+  const project = host.call("project.create", { name: "stack work", repo: "stack" }).id;
+  const A = host.call("thread.create", { project, name: "alpha" }).id;
+  const [{ id: D }] = host.call("session.list", { thread: A });
+  assert.deepEqual(await ask(host.call, A, "compact c1"), { body: "compact c1", before: [] });
+  await until(() => events(f, "end").length === 1, 30000);
+  await settle(() => counted(f).length >= 2);
+  assert.deepEqual(counted(f), [[D, 0, true, null], [D, 1, true, null]]);
+
+  // The run has ended, so the clear retires D at once and starts the successor, whose run-start
+  // collection is its first.
+  const S = host.call("session.clear", { thread: A, session: D, revision: 0, confirm: true, op: "clear-1" }).successor;
+  const done = await settle(() => {
+    const e = host.call("session.operation", { thread: A, op: "clear-1" });
+    return e.start === "started" && e;
+  }, 30000);
+  assert.deepEqual([done?.state, done?.start], ["retired", "started"]);
+  await settle(() => counted(f).some(([s]) => s === S));
+  assert.deepEqual(counted(f).filter(([s]) => s === S)[0], [S, 0, true, null]);
+  const list = () => Object.fromEntries(host.call("session.list", { thread: A }).map((r) => [r.id, r.compactions && { count: r.compactions.count, known: r.compactions.known, reason: r.compactions.reason }]));
+  assert.deepEqual(list(), { [D]: { count: 1, known: true, reason: null }, [S]: { count: 0, known: true, reason: null } });
+
+  // A legacy session (op legacy-3l, from the 3k-3l migration): its store is the thread's, made by
+  // hand as packages/bus/tests/session-compactions.test.mjs does, with a ledger the launcher's
+  // collect wrote.
+  await until(() => events(f, "end").length === 2, 30000);
+  await host.close();
+  const L = "4b1c3a52-9a51-4d7e-8f2e-6c0a1b2d3e4f";
+  const seeded = new Store(f.dataRoot);
+  const at = "2026-10-21T00:00:00.000Z";
+  seeded.run("INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)", L, at, "acme", project, "assistant", f.doc.human, "legacy");
+  seeded.run("INSERT INTO sessions(at,business,thread,id,instance,owner,name,op) SELECT at,business,id,id,instance,owner,'default','legacy-3l' FROM threads WHERE id=?", L);
+  seeded.run("INSERT INTO session_defaults(at,thread,session,by) SELECT at,id,id,'migration-3l' FROM threads WHERE id=?", L);
+  seeded.close();
+  const legacy = sessionStore(f.dataRoot, "acme", L, L, { legacy: true });
+  writeFileSync(join(legacy.session, "pi.jsonl"), line({ type: "session", version: 3, id: "pi", timestamp: "2026-10-22T11:00:00.000Z", cwd: "/w" }) + line({ type: "compaction", id: "c7", parentId: null, timestamp: "2026-10-22T11:07:00.000Z", summary: "s", firstKeptEntryId: "k", tokensBefore: 100 }), { mode: 0o600 });
+  assert.equal(collectCompactions(legacy, "pi").count, 1);
+  host = await start(t, prepared);
+  const [row] = host.call("session.list", { thread: L });
+  assert.equal(row.id, L);
+  assert.deepEqual({ ...row.compactions, at: typeof row.compactions?.at }, { count: 1, known: true, reason: null, at: "string" });
+  assert.deepEqual(list(), { [D]: { count: 1, known: true, reason: null }, [S]: { count: 0, known: true, reason: null } });
+});

packages/cli/tests/fixtures/compaction-adapter.mjs

@@ -0,0 +1,24 @@
+// A stand-in Pi turn for the compaction launch tests (row 270, #1775).
+// session-adapter.mjs's turn, plus a Pi session stream in
+// $MOSAIC_SESSION_DIR: a session header when the stream is new, and for a
+// message "compact <id>" one completed compaction entry with that id, in
+// Pi 0.85.1's shape.
+
+import { appendFileSync, existsSync, readFileSync, writeFileSync } from "node:fs";
+import { join } from "node:path";
+
+const request = process.env.MOSAIC_REQUEST ?? "";
+const body = request.match(/^Message [^\n]*:\n\n([\s\S]*)$/m)?.[1] ?? "";
+const historyFile = join(process.env.MOSAIC_SESSION_DIR, "history");
+const before = existsSync(historyFile) ? readFileSync(historyFile, "utf8").split("\n").filter(Boolean) : [];
+const stream = join(process.env.MOSAIC_SESSION_DIR, "pi.jsonl");
+const line = (e) => `${JSON.stringify(e)}\n`;
+if (!existsSync(stream)) appendFileSync(stream, line({ type: "session", version: 3, id: "pi", timestamp: "2026-10-22T11:00:00.000Z", cwd: process.cwd() }));
+const id = body.match(/^compact (c\d+)$/)?.[1];
+if (id) {
+  const timestamp = new Date(Date.UTC(2026, 9, 22, 11, Number(id.slice(1)))).toISOString();
+  appendFileSync(stream, line({ type: "compaction", id, parentId: null, timestamp, summary: "s", firstKeptEntryId: "k", tokensBefore: 100 }));
+}
+appendFileSync(historyFile, `${body}\n`);
+process.stdout.write(`THREAD ${JSON.stringify({ body, before })}\n`);
+writeFileSync(process.env.MOSAIC_TURN_MARKER, "done\n");

packages/seat/src/compactions.mjs

@@ -20,6 +20,15 @@
 //
 // Claude Code has no pinned compaction format yet (Vikunja 466): a session
 // any Claude run collected for is unknown with harness-unsupported, never 0.
+// So is one whose first collection finds Claude history in claude-projects/
+// from before this ledger existed (row 270 r2, B3).
+//
+// One collection reads at most a budget of stream bytes (row 270 r2, B1).
+// The budget is the system config's compactionReadBytes, which only a human
+// writes (mos-1352, DECISIONS Q222); readBudget() takes the default for a
+// missing or bad value. A collection the budget stops short is unknown with
+// not-collected, and its cursors stop at the last whole line read, so a later
+// collection reads on from there.
 //
 // Ledger lines, each one JSON object with v 1:
 //   { kind: "compaction", entry, completedAt, stream, at }  once per key
@@ -39,6 +48,9 @@ export const PI_VERSION = "0.85.1";
 export const LEDGER = "compactions.jsonl";
 const TAIL = 4096;
 const CHUNK = 1 << 20;
+// BUILD.md (queue-270, round 2) has the measured stream sizes behind the default.
+export const READ_BYTES = 64 * CHUNK;
+const READ_MIN = CHUNK, READ_MAX = 1024 * CHUNK;
 const STREAM = /^[A-Za-z0-9._-]{1,200}\.jsonl$/;
 
 const digest = (buf) => createHash("sha256").update(buf).digest("hex");
@@ -53,6 +65,13 @@ export function storeDir(dataRoot, business, thread, session, { legacy = false }
   return legacy ? dir : join(dir, "sessions", session);
 }
 
+// The budget from the system config's compactionReadBytes: a whole number of
+// bytes from 1 MiB to 1 GiB, or the default for anything else, missing included.
+export function readBudget(value) {
+  return Number.isSafeInteger(value) && value >= READ_MIN && value <= READ_MAX ? value : READ_BYTES;
+}
+
+// Only the short tail reads (at most TAIL bytes) use this.
 function readRange(fd, from, to) {
   const out = Buffer.alloc(to - from);
   let off = 0;
@@ -65,12 +84,15 @@ function readRange(fd, from, to) {
 }
 
 // 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 = {}) {
+// by stream. Reads at most `budget` bytes past the cursors, over all streams.
+// Returns the compactions found past each cursor, the new cursors, and the
+// problems: store-unreadable (a file that can't be read or isn't a Pi
+// session), store-incomplete (a stream that shrank, changed or went away, or
+// a line that isn't JSON) and not-collected (the budget ran out first). A
+// missing directory is empty.
+export function scanPi(dir, cursors = {}, budget = READ_BYTES) {
   const found = [], next = {}, problems = new Set();
+  let left = budget;
   let names;
   try {
     names = readdirSync(dir);
@@ -98,14 +120,10 @@ export function scanPi(dir, cursors = {}) {
         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;
+      const take = (raw) => {
+        const line = raw.toString("utf8");
+        if (!line.trim()) return;
         let e;
         try {
           e = JSON.parse(line);
@@ -114,19 +132,40 @@ export function scanPi(dir, cursors = {}) {
         }
         if (!e || typeof e !== "object" || Array.isArray(e)) {
           problems.add("store-incomplete");
-          continue;
+          return;
         }
         if (!header) {
           header = true;
           if (e.type !== "session") problems.add("store-unreadable");
-          continue;
+          return;
         }
         if (e.type === "compaction" && typeof e.id === "string" && e.id) {
           found.push({ entry: e.id, completedAt: typeof e.timestamp === "string" ? e.timestamp : null, stream: name });
         }
+      };
+      // A CHUNK at a time, up to the budget left. Complete lines only: a line
+      // split across chunks waits in `carry`, and a partial last line waits
+      // for its newline. `end` is past the last whole line.
+      const stop = Math.min(size, from + left);
+      const chunk = Buffer.alloc(Math.min(CHUNK, stop - from));
+      let pos = from, end = from, carry = [];
+      while (pos < stop) {
+        const n = readSync(fd, chunk, 0, Math.min(chunk.length, stop - pos), pos);
+        if (n === 0) break;
+        const view = chunk.subarray(0, n);
+        let i = 0;
+        for (let nl = view.indexOf(0x0a); nl !== -1; nl = view.indexOf(0x0a, i)) {
+          take(carry.length ? Buffer.concat([...carry, view.subarray(i, nl)]) : view.subarray(i, nl));
+          carry = [];
+          i = nl + 1;
+          end = pos + i;
+        }
+        if (i < n) carry.push(Buffer.from(view.subarray(i)));
+        pos += n;
       }
-      const bytes = from + end;
-      next[name] = { bytes, tail: digest(readRange(fd, Math.max(0, bytes - TAIL), bytes)) };
+      left -= pos - from;
+      if (stop < size) problems.add("not-collected");
+      next[name] = { bytes: end, tail: digest(readRange(fd, Math.max(0, end - TAIL), end)) };
     } catch {
       problems.add("store-unreadable");
     } finally {
@@ -169,12 +208,25 @@ export function readLedger(dir) {
     else if (r.kind === "collected") {
       last = r;
       if (r.reason === "store-incomplete") incomplete = true;
-      if (r.harness !== "pi") unsupported = true;
+      if (r.harness !== "pi" || r.reason === "harness-unsupported") unsupported = true;
     } else bad = true;
   }
   return { events, cursors, last, incomplete, unsupported, bad, torn };
 }
 
+// Row 270 r2 (B3): whether the store's claude-projects/ holds anything. A
+// directory that can't be read throws, so the collection is not recorded and
+// the next one checks again.
+function claudeHistory(store) {
+  if (!store.projects) return false;
+  try {
+    return readdirSync(store.projects).length > 0;
+  } catch (e) {
+    if (e.code === "ENOENT") return false;
+    throw e;
+  }
+}
+
 // What session.list shows: { count, known, reason, at }. Known only when the
 // ledger is whole, every collection ran on Pi and none lost entries. Any other
 // answer is unknown with a reason and a lower bound, or null for none: never 0.
@@ -202,12 +254,17 @@ export function compactionCount(dir) {
 // 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 };
+export function collectCompactions(store, harness, { now = () => new Date().toISOString(), budget = READ_BYTES } = {}) {
+  const prior = readLedger(store.dir);
+  const ledger = prior ?? { events: new Map(), cursors: {}, torn: false };
   const at = now();
   const out = [];
-  if (harness === "pi") {
-    const scan = scanPi(store.session, ledger.cursors);
+  if (harness === "pi" && !prior && claudeHistory(store)) {
+    // Claude runs before this ledger: their compactions can't be counted
+    // (Vikunja 466), so the count is unknown for good, as after a Claude run.
+    out.push({ v: 1, kind: "collected", at, known: false, reason: "harness-unsupported", harness, version: PI_VERSION });
+  } else if (harness === "pi") {
+    const scan = scanPi(store.session, ledger.cursors, budget);
     for (const f of scan.found) {
       const key = keyOf(f.entry, f.completedAt);
       if (ledger.events.has(key)) continue;
@@ -219,7 +276,8 @@ export function collectCompactions(store, harness, { now = () => new Date().toIS
       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;
+    // not-collected is last: the next collection reads on.
+    const reason = ["store-incomplete", "store-unreadable", "not-collected"].find((p) => scan.problems.includes(p)) ?? null;
     out.push({ v: 1, kind: "collected", at, known: reason === null, reason, harness, version: PI_VERSION });
   } else {
     out.push({ v: 1, kind: "collected", at, known: false, reason: "harness-unsupported", harness, version: null });

packages/seat/tests/compactions-budget.test.mjs

@@ -0,0 +1,110 @@
+// Row 270 round 2 (264c, #1775): the read budget (B1, duckworth-0223, sage-3031,
+// mos-1352) and Claude history from before the ledger (B3). The streams are Pi
+// 0.85.1's shapes, written by hand as in compactions.test.mjs. A stream over
+// the budget is a sparse file: truncate() makes the hole, so no test writes
+// the bytes it reads past.
+
+import { test } from "node:test";
+import assert from "node:assert/strict";
+import { appendFileSync, chmodSync, mkdtempSync, readFileSync, rmSync, truncateSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { collectCompactions, LEDGER, READ_BYTES, readBudget } from "../src/compactions.mjs";
+import { sessionStore } from "../src/session.mjs";
+
+const MiB = 1 << 20;
+
+function setup(t) {
+  const root = mkdtempSync(join(tmpdir(), "seat-compactions-budget-"));
+  t.after(() => rmSync(root, { recursive: true, force: true }));
+  const store = sessionStore(root, "acme", "th1", "s1");
+  let tick = 0;
+  const now = () => new Date(Date.UTC(2026, 9, 22, 12, 0, tick++)).toISOString();
+  const collect = (harness = "pi", budget = undefined) => collectCompactions(store, harness, { now, ...(budget === undefined ? {} : { budget }) });
+  const records = () => readFileSync(join(store.dir, LEDGER), "utf8").trim().split("\n").map((l) => JSON.parse(l));
+  return { root, store, collect, records };
+}
+
+const header = (id) => ({ type: "session", version: 3, id, timestamp: "2026-10-22T11:00:00.000Z", cwd: "/w" });
+const compaction = (id, timestamp) => ({ type: "compaction", id, parentId: null, timestamp, summary: "s", firstKeptEntryId: "k", tokensBefore: 100 });
+const line = (e) => `${JSON.stringify(e)}\n`;
+const at = (n) => `2026-10-22T12:00:0${n}.000Z`;
+
+test("the budget is the setting's whole number of bytes from 1 MiB to 1 GiB, and the default for anything else", () => {
+  assert.equal(READ_BYTES, 64 * MiB);
+  for (const v of [MiB, 8 * MiB, 1024 * MiB]) assert.equal(readBudget(v), v);
+  for (const v of [undefined, null, "8388608", 8.5 * MiB, -1, 0, MiB - 1, 1024 * MiB + 1, NaN, Infinity, {}, [MiB], true]) {
+    assert.equal(readBudget(v), READ_BYTES, JSON.stringify(v) ?? String(v));
+  }
+});
+
+test("a stream past the default budget is not-collected with a lower bound, and a later collection catches up", (t) => {
+  const f = setup(t);
+  const path = join(f.store.session, "a.jsonl");
+  const whole = line(header("a")) + line(compaction("c1", "2026-10-22T11:01:00.000Z"));
+  writeFileSync(path, whole);
+  // A hole past the default budget, with no newline in it: a line still being written.
+  truncateSync(path, READ_BYTES + 2 * MiB);
+  assert.deepEqual(f.collect(), { count: 1, known: false, reason: "not-collected", at: at(0) });
+  // The cursor stops at the last whole line read.
+  assert.deepEqual(f.records().filter((r) => r.kind === "cursor").map((r) => r.bytes), [Buffer.byteLength(whole)]);
+  // The long line ends up as c2: the bytes past the cursor are read again, from the cursor.
+  truncateSync(path, Buffer.byteLength(whole));
+  appendFileSync(path, line(compaction("c2", "2026-10-22T11:02:00.000Z")));
+  assert.deepEqual(f.collect(), { count: 2, known: true, reason: null, at: at(1) });
+});
+
+test("a budget smaller than the streams counts over several collections, once each, and is known when it catches up", (t) => {
+  const f = setup(t);
+  const h = Buffer.byteLength(line(header("a"))), c = Buffer.byteLength(line(compaction("c1", "2026-10-22T11:01:00.000Z")));
+  writeFileSync(join(f.store.session, "a.jsonl"), line(header("a")) + line(compaction("c1", "2026-10-22T11:01:00.000Z")) + line(compaction("c2", "2026-10-22T11:02:00.000Z")));
+  writeFileSync(join(f.store.session, "b.jsonl"), line(header("b")) + line(compaction("c3", "2026-10-22T11:03:00.000Z")));
+  // One byte past a's first compaction: a stops there and b isn't reached; then all of c2 and b's
+  // header; then c3.
+  const budget = h + c + 1;
+  assert.deepEqual(f.collect("pi", budget), { count: 1, known: false, reason: "not-collected", at: at(0) });
+  assert.deepEqual(f.collect("pi", budget), { count: 2, known: false, reason: "not-collected", at: at(1) });
+  assert.deepEqual(f.collect("pi", budget), { count: 3, known: true, reason: null, at: at(2) });
+  assert.deepEqual(f.records().filter((r) => r.kind === "compaction").map((r) => [r.entry, r.stream]), [["c1", "a.jsonl"], ["c2", "a.jsonl"], ["c3", "b.jsonl"]]);
+});
+
+test("a line split across reads counts once", (t) => {
+  const f = setup(t);
+  // A message that ends just short of the first read's end, so c1 crosses it.
+  const first = line(header("a"));
+  const pad = { type: "message", id: "m1", parentId: null, timestamp: "2026-10-22T11:00:01.000Z", message: { role: "user", content: "" } };
+  pad.message.content = "x".repeat(MiB - 100 - Buffer.byteLength(first) - Buffer.byteLength(line(pad)));
+  writeFileSync(join(f.store.session, "a.jsonl"), first + line(pad) + line(compaction("c1", "2026-10-22T11:01:00.000Z")));
+  assert.equal(Buffer.byteLength(first + line(pad)), MiB - 100);
+  assert.deepEqual(f.collect(), { count: 1, known: true, reason: null, at: at(0) });
+});
+
+// B3 (Vikunja 466): Claude Code runs before this ledger left history in claude-projects/ that
+// can't be counted. The first collection is harness-unsupported, and it stays so.
+test("Claude history from before the ledger makes the first Pi collection harness-unsupported, for good", (t) => {
+  const f = setup(t);
+  writeFileSync(join(f.store.projects, "old.jsonl"), "{}\n");
+  writeFileSync(join(f.store.session, "a.jsonl"), line(header("a")) + line(compaction("c1", "2026-10-22T11:01:00.000Z")));
+  assert.deepEqual(f.collect(), { count: null, known: false, reason: "harness-unsupported", at: at(0) });
+  assert.deepEqual(f.collect(), { count: 1, known: false, reason: "harness-unsupported", at: at(1) });
+});
+
+test("an empty claude-projects/ is no Claude history, and one filled after the ledger exists isn't checked", (t) => {
+  const f = setup(t);
+  writeFileSync(join(f.store.session, "a.jsonl"), line(header("a")) + line(compaction("c1", "2026-10-22T11:01:00.000Z")));
+  assert.deepEqual(f.collect(), { count: 1, known: true, reason: null, at: at(0) });
+  // A Claude run after the ledger exists collects for itself (compactions.test.mjs).
+  writeFileSync(join(f.store.projects, "later.jsonl"), "{}\n");
+  assert.deepEqual(f.collect(), { count: 1, known: true, reason: null, at: at(1) });
+});
+
+test("a claude-projects/ that can't be read records nothing, and the next collection checks again", (t) => {
+  const f = setup(t);
+  writeFileSync(join(f.store.projects, "old.jsonl"), "{}\n");
+  chmodSync(f.store.projects, 0o000);
+  t.after(() => chmodSync(f.store.projects, 0o700));
+  assert.throws(() => f.collect(), { code: "EACCES" });
+  assert.throws(() => readFileSync(join(f.store.dir, LEDGER)), { code: "ENOENT" });
+  chmodSync(f.store.projects, 0o700);
+  assert.deepEqual(f.collect(), { count: null, known: false, reason: "harness-unsupported", at: at(1) });
+});

scripts/mosaic-config.mjs

@@ -294,9 +294,13 @@ async function validate(document, file) {
   if (!isPlainObject(document)) {
     fail(2, "configuration must be a JSON object");
   }
+  // Row 270 r2 (mos-1352, DECISIONS Q222): compactionReadBytes, optional and human-only, passes
+  // on as written; the launcher takes the default for a missing or bad value
+  // (packages/seat/src/compactions.mjs readBudget). The first list stays as it was:
+  // packages/cli/tests/stack-release-fixture.mjs edits it to make an older loader.
   rejectUnknownKeys(
     document,
-    ["configVersion", "environment", "dataRoot", "execution", "console", "federation"],
+    ["configVersion", "environment", "dataRoot", "execution", "console", "federation"].concat(["compactionReadBytes"]),
     "configuration",
   );
 
@@ -349,6 +353,7 @@ async function validate(document, file) {
     },
     ...(consoleKey === undefined ? {} : { console: consoleKey }),
     ...(federationKey === undefined ? {} : { federation: federationKey }),
+    ...(Object.hasOwn(document, "compactionReadBytes") ? { compactionReadBytes: document.compactionReadBytes } : {}),
   };
 }
 

scripts/test-config.sh

@@ -263,6 +263,16 @@ EXPECTED="$(printf '{\n  "configVersion": 1,\n  "environment": "development",\n
 [ "$(MOSAIC_CONFIG="$SANDBOX/no-console.json" timeout 20 $CONFIG_OP validate)" = "$EXPECTED" ]
 check "config without console loads with unchanged output" $?
 
+# --- Row 270 r2 (mos-1352, DECISIONS Q222): the optional compactionReadBytes key ---
+# The loader passes it on as written, a bad value too: the launcher takes the default for a
+# missing or bad one (packages/seat/src/compactions.mjs readBudget), so it never refuses a config.
+for v in 8388608 '"x"'; do
+    cfg budget.json '{"configVersion":1,"environment":"development","dataRoot":"'$DATA_ROOT'","execution":{"backend":"docker","provider":"zai","model":"m"},"compactionReadBytes":'"$v"'}'
+    OUT="$(MOSAIC_CONFIG="$SANDBOX/budget.json" timeout 20 $CONFIG_OP validate 2>&1)"
+    [ $? -eq 0 ] && printf '%s' "$OUT" | grep -qxF "  \"compactionReadBytes\": $v"
+    check "compactionReadBytes $v loads and is in the output as written" $?
+done
+
 # --- validation must not modify the file ---
 SUM_INVALID_BEFORE=$(sha256sum "$SANDBOX/invalid.json" | cut -d' ' -f1)
 MOSAIC_CONFIG="$SANDBOX/invalid.json" $CONFIG_OP validate >/dev/null 2>&1