gizmoduck: row 270 (264c, #1775) round 3 for duckworth-0226 B4 (sage-3038): one collection opens at most the system config's compactionStreams stream files in name order (default 1024, 32 to 65536, a bad value takes the default and logs), not-collected past it with no cursor for an unopened stream; the missing-stream check uses a Set; readStreams, cap and launcher tests; mutants M22-M24, M10 anchor moved; docs/TOOLS.md has both keys and C1 (N4); BUILD.md round 3 with stream counts, the 231 KB bound, C1 and N5; not run

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

@@ -361,3 +361,107 @@ fix or a second red goes to Sage and Mos first.
   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.
+
+## Round 3 (duckworth-0226 B4, sage-3038)
+
+Round 2 at 62a279cf3 got CHANGES on B4. B1-B3 were met, and Duckworth accepted the schema change.
+
+### B4: a cap on the streams one collection opens
+
+Before this round, one collection opened every `*.jsonl` in the session directory. It also checked
+each ledger cursor with `names.includes`, which is O(cursors × names), and wrote a cursor line for
+every stream, empty ones included. The agent can write its rw session directory (sandbox.mjs:124),
+and nothing limits its file count. So 100,000 empty files meant about 22 MB of cursor lines at one
+run end, then about 10^10 string compares on the launcher's loop at the next.
+
+Now:
+
+- scanPi (compactions.mjs:123-125) sorts the `*.jsonl` names and opens at most `streams` of them.
+  When there are more, the collection adds not-collected. The streams it didn't open get no cursor
+  line. A stream that already has a cursor keeps its old one.
+- The missing-stream check uses a Set (compactions.mjs:121-122).
+- The cap is a setting, per sage-3038 (the same reason as mos-1352 and Q222): top-level
+  `compactionStreams` in the instance system config. It follows compactionReadBytes. It is
+  human-only. `readStreams()` accepts a safe integer from 32 to 65536 and returns the default, 1024,
+  for anything else. The launcher reads it once and logs one `system config:` line for a bad value.
+  mosaic-config.mjs adds the key to the same `.concat` and passes it on verbatim. The fixture literal
+  at stack-release-fixture.mjs:300 still matches.
+- Name order stays sorted, so a collection under a larger cap, or with fewer files, catches up.
+  Under the same cap and the same files, each collection opens the same streams and adds no cursor
+  lines for streams that haven't moved.
+
+Default, from stat only (`find ~/.pi/agent/sessions -name '*.jsonl'`, counted per directory, no
+content read). Each of these directories holds every session for one cwd, so a single session's
+count is at most these:
+
+| directories | p50 | p90 | p99 | max |
+|---|---|---|---|---|
+| 40 | 3 | 19 | 27 | 27 |
+
+1024 is about 38 times the max. The cursor lines one collection writes are bounded by 1024 × 226 B,
+about 231 KB. At the setting's 65536 maximum the bound is about 14.8 MB.
+
+### N4: docs/TOOLS.md
+
+docs/TOOLS.md has a paragraph on both keys, after the `config-console-key` paragraph. It gives the
+range, the default, that only a human writes them, the fallback with its log line, not-collected,
+and C1. No test reads TOOLS.md. stack-release.test.mjs:78 only names it inside a message string.
+
+### C1 (duckworth-0226, sage-3038)
+
+Any re-pin or rollback plan whose OLD copy is before row 270 removes both `compactionReadBytes` and
+`compactionStreams` from config.json first. A pre-270 loader refuses either key. stack-release.sh
+config_checks (:1079-1097) runs the target loader only when there is a `console` key, so the release
+doesn't catch this. Sage is drafting the config_checks follow-up row.
+
+### Tests
+
+- seat compactions-budget.test.mjs:114-120: readStreams over good and bad values.
+- :122-134: four streams under a cap of 3. Both collections are not-collected, and the cursors are
+  only a, b and c, with no new cursor lines on the second. A cap of 4 then reads count 4, known.
+- :136-143: 1025 empty streams under the default cap. The result is not-collected with 1024 cursor
+  lines, the last one e1023.
+- cli compactions-launch.test.mjs:264-298: compactionStreams 32 with 34 streams. pi.jsonl sorts
+  last, so its c0 isn't opened and both collections are `[D,null,false,not-collected]`. Then `"x"`
+  logs the fallback, and the next run's collections read `[D,1,true,null]`.
+- scripts/test-config.sh: 2048 and `"x"` load and pass through, as for compactionReadBytes.
+
+### Mutants
+
+M10's anchor moved to the Set line. The mutant is the same, removing the missing-stream check. Every
+anchor in mutants.py matches once (`check`). Round 3 shifted round 2's kill lines: seat budget
+:37/:48/:79/:88/:89 are now :39/:50/:81/:90/:91, and cli launch :182/:197/:248 are now
+:183/:198/:249. mutants.py K has the current lines. The launcher call sites are now
+launcher.mjs:507 and :648.
+
+| id | change | kill |
+|---|---|---|
+| M22 | no cap, `for (const name of visit)` | seat compactions-budget.test.mjs:127, all four are opened: count 4 known; :141 fails the same way |
+| M23 | `readStreams` returns `value ?? STREAMS` | :118, `"2048"` comes back as the string |
+| M24 | the launcher ignores the setting, `streams = STREAMS` | cli compactions-launch.test.mjs:285, all 34 open: `[D,1,true,null]` twice |
+
+sha256 at this round's tip, to check each restore against. These replace round 2's table:
+
+```
+b5a078a096e447c4afe15812696d1f2f8c3f2b7665868c02e61a4e1edc8cd874  packages/seat/src/compactions.mjs
+d7cdc8a17e337e00dfe2d1bf6249cc2c2e696b16c5b426b2fe616089e5559509  packages/bus/src/broker.mjs
+21a7e910d81848aeb99036028d5b87354a2a515f3642779105c881b43a8b6e05  packages/cli/src/launcher.mjs
+5ae423438a178b1c8eac3b8ce0e0c22f5bb7caebb9011ce7098a36d95b53f1ca  packages/bus/src/runtime.mjs
+```
+
+### Run plan
+
+This is round 2's plan with M1-M24. M22 and M23 run seat compactions-budget.test.mjs (120 s). M24
+runs cli compactions-launch.test.mjs (600 s). Per sage-3038, the slot goes to Megavolt's 207, then
+203, then this row's combined 270+271 gate after Duckworth approves r3.
+
+### Residuals
+
+- N5. readdirSync still lists the whole session directory, and the names are sorted before the
+  cap. Both are O(files) in memory and time on the launcher. The cap bounds opens, reads and ledger
+  lines, not the listing. Bounding the listing needs opendirSync with a stop. That would lose name
+  order, and name order is what lets a later collection catch up. Accepted for this row
+  (duckworth-0226).
+- The ledger's cursor map still grows when the agent deletes and recreates streams, by at most the
+  cap per collection. Each deletion of a stream with a cursor is store-incomplete for good. This is
+  the known limit that the bus reads the ledger whole (round 2).

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

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 # 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
+# (duckworth-0223 B1-B3, sage-3030, sage-3031, mos-1352); M22-M24 are round 3's (duckworth-0226
+# B4, sage-3038). Round 3's kill lines are at round 3's tests. `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
@@ -28,7 +29,8 @@ M = {
  "M7": (C, '  if (ledger.incomplete) return unknown("store-incomplete", count, at);\n', ""),
  "M8": (C, '  if (ledger.unsupported) return unknown("harness-unsupported", count, at);\n', ""),
  "M9": (C, "({ count: count || null, known: false, reason, at })", "({ count, known: false, reason, at })"),
- "M10": (C, '  for (const stream of Object.keys(cursors)) if (!names.includes(stream)) problems.add("store-incomplete");\n', ""),
+ # Round 3 moved M10's anchor to the Set (duckworth-0226 B4); the mutant is the same.
+ "M10": (C, '  for (const stream of Object.keys(cursors)) if (!present.has(stream)) problems.add("store-incomplete");\n', ""),
  "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 };"),
@@ -50,6 +52,12 @@ M = {
  "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;'),
+ # B4: the stream cap and its setting.
+ "M22": (C, '  if (visit.length > streams) problems.add("not-collected");\n  for (const name of visit.slice(0, streams)) {',
+            '  for (const name of visit) {'),
+ "M23": (C, "  return Number.isSafeInteger(value) && value >= STREAMS_MIN && value <= STREAMS_MAX ? value : STREAMS;",
+            "  return value ?? STREAMS;"),
+ "M24": (L, "  const streams = readStreams(system?.compactionStreams);", "  const streams = STREAMS;"),
 }
 K = {
  "M1": (ST, "a compaction copied into a branched stream counts once, under the stream it was first seen in", 64,
@@ -73,24 +81,30 @@ 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,
+ "M13": (CT, "a run collects compactions at its start and its end, and the count holds across a resume and a host restart", 183,
          "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,
+ "M14": (CT, "a run collects compactions at its start and its end, and the count holds across a resume and a host restart", 183,
          "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,
+ "M15": (CT, "a run collects compactions at its start and its end, and the count holds across a resume and a host restart", 198,
+         "session.list's compactions is null: {at: 'undefined'} and no count; the clear test's :323 and the legacy :343 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", 50,
          "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,
+ "M17": (SB, "the budget is the setting's whole number of bytes from 1 MiB to 1 GiB, and the default for anything else", 39,
          "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,
+ "M18": (CT, "the read budget is the system config's compactionReadBytes, and a bad value takes the default", 249,
          "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,
+ "M19": (SB, "a line split across reads counts once", 81,
          "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,
+ "M20": (SB, "Claude history from before the ledger makes the first Pi collection harness-unsupported, for good", 90,
          "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,
+ "M21": (SB, "Claude history from before the ledger makes the first Pi collection harness-unsupported, for good", 91,
          "a Pi collected record isn't sticky: the second reads {count 1, known true}"),
+ "M22": (SB, "more streams than the cap is not-collected, with cursors only for the streams opened, and a larger cap catches up", 127,
+         "all four streams are opened: {count 4, known true, reason null}; the default-cap test's :141 fails the same way (1025 cursors)"),
+ "M23": (SB, "the stream cap is the setting's whole number from 32 to 65536, and the default for anything else", 118,
+         "readStreams(null) is the default, but readStreams('2048') is the string"),
+ "M24": (CT, "the stream cap is the system config's compactionStreams, and a bad value takes the default", 285,
+         "the cap of 32 is ignored, all 34 streams are opened: [D,1,true,null] twice"),
 }
 cmd = sys.argv[1]
 if cmd == "list":

docs/TOOLS.md

@@ -638,6 +638,18 @@ before row 192 refuses the key). The message says to remove the key first. So a
 window re-pins to row 192 or later before it writes the key, and a rollback past
 row 192 removes the key first.
 
+Two more optional system config keys bound the host's compaction collections
+(row 270, #1775, DECISIONS Q222). Only a human writes them, like `console`.
+`compactionReadBytes` is the stream bytes one collection reads, a whole number
+from 1048576 to 1073741824, default 67108864 (64 MiB). `compactionStreams` is the
+stream files one collection opens, a whole number from 32 to 65536, default 1024.
+The loader passes either on as written. The launcher takes the default for a
+missing or bad value and logs one `system config:` line. A collection that hits
+either bound counts as `not-collected`, and the next one reads on. A copy before
+row 270 refuses both keys, and the release's config check doesn't catch that
+(it runs the target's loader only for a `console` key). So any re-pin or
+rollback plan whose OLD copy is before row 270 removes both keys first.
+
 #### A target past c216f47db: the PM after the harness kind
 
 From c216f47db on, the bus launches the PM only with a `harness` kind in the

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, READ_BYTES, readBudget } from "../../seat/src/compactions.mjs";
+import { collectCompactions, READ_BYTES, readBudget, readStreams, STREAMS } 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";
@@ -298,10 +298,16 @@ export function createLauncher({
   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}`);
   }
+  // Row 270 r3 (B4, sage-3038): and opens at most compactionStreams stream
+  // files, the same kind of setting.
+  const streams = readStreams(system?.compactionStreams);
+  if (system?.compactionStreams !== undefined && streams !== system.compactionStreams) {
+    log(`system config: compactionStreams ${JSON.stringify(system.compactionStreams)} isn't a whole number from 32 to 65536; compaction collections open up to the default ${STREAMS} streams`);
+  }
   const compactions = (store, harness, { instance, thread, session, run }) => {
     let c;
     try {
-      c = collectCompactions(store, harness, { budget });
+      c = collectCompactions(store, harness, { budget, streams });
     } catch (e) {
       log(`run ${run}: compactions: ${e.message}`);
       c = { count: null, known: false, reason: "store-unreadable" };

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

@@ -7,7 +7,8 @@
 //
 // 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.
+// successor, and a legacy session read through the runtime's reader. Round 3
+// (duckworth-0226 B4, sage-3038): the stream cap from the system config.
 //
 // 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
@@ -260,6 +261,42 @@ test("REQ-THREAD-2 T3 the read budget is the system config's compactionReadBytes
   assert.deepEqual(counted(f).slice(2), [[D, 1, true, null], [D, 1, true, null]]);
 });
 
+test("REQ-THREAD-2 T3 the stream cap is the system config's compactionStreams, 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.compactionStreams = v;
+    writeFileSync(f.config, JSON.stringify(doc));
+  };
+  setting(32);
+  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 });
+  // 33 empty streams that sort before pi.jsonl: 34 in all, past a cap of 32, so pi.jsonl and its
+  // c0 aren't opened. The default cap opens all 34.
+  const store = sessionStore(f.dataRoot, "acme", A, D);
+  for (let i = 0; i < 33; i++) writeFileSync(join(store.session, `e${String(i).padStart(2, "0")}.jsonl`), "", { mode: 0o600 });
+  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 });
+  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, null, false, "not-collected"], [D, null, 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: compactionStreams "x" isn\'t a whole number from 32 to 65536; compaction collections open up to the default 1024 streams',
+  ]);
+  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;

packages/seat/src/compactions.mjs

@@ -30,6 +30,13 @@
 // not-collected, and its cursors stop at the last whole line read, so a later
 // collection reads on from there.
 //
+// One collection opens at most a cap of stream files, in name order (row 270
+// r3, B4). The cap is the system config's compactionStreams, human-only too
+// (sage-3038); readStreams() takes the default for a missing or bad value.
+// Past the cap a collection is not-collected, and the streams it didn't open
+// get no cursor line, so neither a collection's work nor the ledger lines it
+// writes grow with the number of files the agent makes.
+//
 // 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`
@@ -51,6 +58,9 @@ 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;
+// BUILD.md (queue-270, round 3) has the measured stream counts behind the default.
+export const STREAMS = 1024;
+const STREAMS_MIN = 32, STREAMS_MAX = 65536;
 const STREAM = /^[A-Za-z0-9._-]{1,200}\.jsonl$/;
 
 const digest = (buf) => createHash("sha256").update(buf).digest("hex");
@@ -71,6 +81,12 @@ export function readBudget(value) {
   return Number.isSafeInteger(value) && value >= READ_MIN && value <= READ_MAX ? value : READ_BYTES;
 }
 
+// The stream cap from the system config's compactionStreams: a whole number
+// from 32 to 65536, or the default for anything else, missing included.
+export function readStreams(value) {
+  return Number.isSafeInteger(value) && value >= STREAMS_MIN && value <= STREAMS_MAX ? value : STREAMS;
+}
+
 // Only the short tail reads (at most TAIL bytes) use this.
 function readRange(fd, from, to) {
   const out = Buffer.alloc(to - from);
@@ -84,13 +100,15 @@ function readRange(fd, from, to) {
 }
 
 // One collection over a Pi session directory. `cursors` are the ledger's,
-// by stream. Reads at most `budget` bytes past the cursors, over all streams.
+// by stream. Opens at most `streams` streams, in name order, and reads at
+// most `budget` bytes past their cursors, over all of them.
 // Returns the compactions found past each cursor, the new cursors, and the
 // problems: store-unreadable (a file that can't be read or isn't a Pi
 // session), store-incomplete (a stream that shrank, changed or went away, or
-// a line that isn't JSON) and not-collected (the budget ran out first). A
+// a line that isn't JSON) and not-collected (the budget or the stream cap
+// ran out first). A
 // missing directory is empty.
-export function scanPi(dir, cursors = {}, budget = READ_BYTES) {
+export function scanPi(dir, cursors = {}, budget = READ_BYTES, streams = STREAMS) {
   const found = [], next = {}, problems = new Set();
   let left = budget;
   let names;
@@ -100,8 +118,11 @@ export function scanPi(dir, cursors = {}, budget = READ_BYTES) {
     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 present = new Set(names);
+  for (const stream of Object.keys(cursors)) if (!present.has(stream)) problems.add("store-incomplete");
+  const visit = names.filter((n) => n.endsWith(".jsonl")).sort();
+  if (visit.length > streams) problems.add("not-collected");
+  for (const name of visit.slice(0, streams)) {
     const path = join(dir, name);
     let fd;
     try {
@@ -254,7 +275,7 @@ 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(), budget = READ_BYTES } = {}) {
+export function collectCompactions(store, harness, { now = () => new Date().toISOString(), budget = READ_BYTES, streams = STREAMS } = {}) {
   const prior = readLedger(store.dir);
   const ledger = prior ?? { events: new Map(), cursors: {}, torn: false };
   const at = now();
@@ -264,7 +285,7 @@ export function collectCompactions(store, harness, { now = () => new Date().toIS
     // (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);
+    const scan = scanPi(store.session, ledger.cursors, budget, streams);
     for (const f of scan.found) {
       const key = keyOf(f.entry, f.completedAt);
       if (ledger.events.has(key)) continue;

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

@@ -1,5 +1,6 @@
 // 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
+// mos-1352) and Claude history from before the ledger (B3). Round 3: the cap on
+// streams one collection opens (B4, duckworth-0226). 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.
@@ -9,7 +10,7 @@ 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 { collectCompactions, LEDGER, READ_BYTES, readBudget, readStreams, STREAMS } from "../src/compactions.mjs";
 import { sessionStore } from "../src/session.mjs";
 
 const MiB = 1 << 20;
@@ -20,7 +21,8 @@ function setup(t) {
   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 collect = (harness = "pi", budget = undefined, streams = undefined) =>
+    collectCompactions(store, harness, { now, ...(budget === undefined ? {} : { budget }), ...(streams === undefined ? {} : { streams }) });
   const records = () => readFileSync(join(store.dir, LEDGER), "utf8").trim().split("\n").map((l) => JSON.parse(l));
   return { root, store, collect, records };
 }
@@ -108,3 +110,34 @@ test("a claude-projects/ that can't be read records nothing, and the next collec
   chmodSync(f.store.projects, 0o700);
   assert.deepEqual(f.collect(), { count: null, known: false, reason: "harness-unsupported", at: at(1) });
 });
+
+test("the stream cap is the setting's whole number from 32 to 65536, and the default for anything else", () => {
+  assert.equal(STREAMS, 1024);
+  for (const v of [32, 2048, 65536]) assert.equal(readStreams(v), v);
+  for (const v of [undefined, null, "2048", 40.5, -1, 0, 31, 65537, NaN, Infinity, {}, [64], true]) {
+    assert.equal(readStreams(v), STREAMS, JSON.stringify(v) ?? String(v));
+  }
+});
+
+test("more streams than the cap is not-collected, with cursors only for the streams opened, and a larger cap catches up", (t) => {
+  const f = setup(t);
+  const names = ["a", "b", "c", "d"];
+  names.forEach((n, i) => writeFileSync(join(f.store.session, `${n}.jsonl`), line(header(n)) + line(compaction(`c${n}`, `2026-10-22T11:0${i}:00.000Z`))));
+  const cursors = () => f.records().filter((r) => r.kind === "cursor").map((r) => r.stream);
+  assert.deepEqual(f.collect("pi", undefined, 3), { count: 3, known: false, reason: "not-collected", at: at(0) });
+  assert.deepEqual(cursors(), ["a.jsonl", "b.jsonl", "c.jsonl"]);
+  // The same cap again opens the same three, which haven't moved: no new lines but the collected one.
+  assert.deepEqual(f.collect("pi", undefined, 3), { count: 3, known: false, reason: "not-collected", at: at(1) });
+  assert.deepEqual(cursors(), ["a.jsonl", "b.jsonl", "c.jsonl"]);
+  assert.deepEqual(f.collect("pi", undefined, 4), { count: 4, known: true, reason: null, at: at(2) });
+  assert.deepEqual(cursors(), ["a.jsonl", "b.jsonl", "c.jsonl", "d.jsonl"]);
+});
+
+test("the default cap opens STREAMS streams and writes no more cursor lines than that", (t) => {
+  const f = setup(t);
+  for (let i = 0; i <= STREAMS; i++) writeFileSync(join(f.store.session, `e${String(i).padStart(4, "0")}.jsonl`), "");
+  assert.deepEqual(f.collect(), { count: null, known: false, reason: "not-collected", at: at(0) });
+  const cursors = f.records().filter((r) => r.kind === "cursor");
+  assert.equal(cursors.length, STREAMS);
+  assert.equal(cursors.at(-1).stream, "e1023.jsonl");
+});

scripts/mosaic-config.mjs

@@ -241,13 +241,14 @@ 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:
+  // Row 270 r2 and r3 (mos-1352, DECISIONS Q222, sage-3038): compactionReadBytes and
+  // compactionStreams, optional and human-only, pass on as written; the launcher takes the
+  // default for a missing or bad value (packages/seat/src/compactions.mjs readBudget and
+  // readStreams). 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"].concat(["compactionReadBytes"]),
+    ["configVersion", "environment", "dataRoot", "execution", "console"].concat(["compactionReadBytes", "compactionStreams"]),
     "configuration",
   );
 
@@ -297,6 +298,7 @@ async function validate(document, file) {
     },
     ...(consoleKey === undefined ? {} : { console: consoleKey }),
     ...(Object.hasOwn(document, "compactionReadBytes") ? { compactionReadBytes: document.compactionReadBytes } : {}),
+    ...(Object.hasOwn(document, "compactionStreams") ? { compactionStreams: document.compactionStreams } : {}),
   };
 }
 

scripts/test-config.sh

@@ -206,6 +206,13 @@ for v in 8388608 '"x"'; do
     [ $? -eq 0 ] && printf '%s' "$OUT" | grep -qxF "  \"compactionReadBytes\": $v"
     check "compactionReadBytes $v loads and is in the output as written" $?
 done
+# Row 270 r3 (sage-3038): compactionStreams the same way (readStreams).
+for v in 2048 '"x"'; do
+    cfg streams.json '{"configVersion":1,"environment":"development","dataRoot":"'$DATA_ROOT'","execution":{"backend":"docker","provider":"zai","model":"m"},"compactionStreams":'"$v"'}'
+    OUT="$(MOSAIC_CONFIG="$SANDBOX/streams.json" timeout 20 $CONFIG_OP validate 2>&1)"
+    [ $? -eq 0 ] && printf '%s' "$OUT" | grep -qxF "  \"compactionStreams\": $v"
+    check "compactionStreams $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)