row 271 (264d, #1776): schema step 3l to 3m, session_ops and message_cancels with 7 triggers; successor id pinned as a lowercase version 8 UUID for Huey (gizmoduck-0274)

A halt inside the op reaches the bus's own fence first (runtime.mjs:564-566,
processingWork at :352-356), so the launcher refused limit-check-failed
(launcher.mjs:389-391) before its recheck admission() at :397. First-run red of
a new test on 6d9ab3503 early signal, placed under mos-1272.

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

@@ -0,0 +1,449 @@
+# Row 269 build notes (Gizmoduck)
+
+264b, T1 item 7: per-session native store, workspace and single-writer fence. Gitea #1774. Brief:
+docs/plans/2026-10-21_t1-sessions.md (blob ff0be0c79), slices in agents/sage/work/queue-264/RULINGS.md
+(5f5b51dd6). Reviewer Duckworth, gate owner Sage. Plan gizmoduck-0228, accepted with rulings in
+sage-2707. Branch gizmoduck/row269 from refactor aca16404c. Row 253 lands first, and this branch is
+rebased onto its tip, under a new name, before the gate.
+
+Claim: rev 3217, op gizmoduck-0230, 2026-10-22 14:29:42Z. That was 21 s after Gander's slot opened
+at 14:29:21Z (sage-2710). Reported in gizmoduck-0231 and not reverted, since a revert would be a
+second queue write.
+
+## D9 consumer audit, sessions.json (sage-2707 Q3)
+
+Q3 allows `session` as an optional field with no version bump, if every sessions.json reader at
+84f53540d selects by run and not by business, instance or thread. Every reader at 84f53540d:
+
+| Reader | file:line at 84f53540d | Selects by |
+| --- | --- | --- |
+| `readSessions` (the only parser) | packages/seat/src/session.mjs:232-243 | nothing. Refuses any `sessionsVersion` but 1 (:241) and returns every entry with `live` |
+| `stopSession` (`mosaic stop <run>`) | session.mjs:249-250 | run (`s.run === run`) |
+| `launches list` | packages/cli/src/cli.mjs:526-530 | nothing. Prints every entry, and `--json` dumps them whole |
+| host start, `leftover` | packages/cli/src/launcher.mjs:235 | business, then `recover()` works one entry (one run) at a time (:955-972) |
+| runtime launch settings | packages/bus/src/runtime.mjs:149 -> packages/business/src/launch-settings.mjs:166 -> packages/business/src/launch.mjs:116-135 (`usageCounts`) | business. Counts entries per provider and model |
+| Console settings (C1, duckworth-0150) | packages/webui/src/reads.mjs:234 -> the same launchSettings `read()` -> readSessions | business, through usageCounts. Any throw becomes `launch: {refused: 'not-configured'}` |
+| stack-release R guard (C1) | scripts/stack-release.sh:2091-2096, its own JSON.parse, no version check | nothing. Takes `pid` and `runnerPid` of every entry in `sessions` and fails `sessions-live` if one is alive in /proc |
+
+Not a reader: stack-release.sh:2199-2204 checks only that no sessions.json exists. packages/ledger
+readSessions (ledger.mjs:146) is a different function that reads seat directories.
+launch-settings `current()` (:129) doesn't call `sessions()`, so the limits the bus enforces don't
+read the file; only the `limits` verb's `read()` (:166, runtime.mjs:292) does.
+
+Tests that read the file: packages/cli/tests/launcher.test.mjs (`readSessions` at :164 through
+:1097, each by run, by instance or whole) and packages/cli/tests/host.test.mjs:830 (file absent).
+
+Not a file reader: `Registry` (session.mjs:192-230). Its list starts empty (:193), and the constructor
+writes it over the file (:198). `running(business, instance, thread)` at :217 is the one key on the
+thread, but it reads only the in-memory list this host added (launcher.mjs:529, removed at :553).
+Its callers are launcher.mjs:289 (check) and the thread pass. This row changes that key (see Design).
+
+Result: no file reader keys on the thread. Q3's condition holds as worded.
+
+One more finding sends me to v2 anyway. `recover()` rebinds a leftover run with
+`{business, role, run, harness, pid, startTime, thread}` and no `session` (launcher.mjs:964).
+The broker compares that record's body with the recorded `session.launched` body, and the body
+carries `session` whenever the launch record had one (broker.mjs:945). A mismatch refuses
+`launch-record-mismatch` (broker.mjs:952). Each new thread run binds with its session (Design), so
+after a rollback to 264a, a 264a host that finds a 264b entry logs "not ended, bindLaunch refused:
+launch-record-mismatch" (launcher.mjs:968), and the run's claim stays held. The host has
+already killed the process (:956-962), but it can't end the run. With v1 and an optional field,
+that failure is quiet.
+
+Decision: sessions.json goes to v2 (`SESSIONS_VERSION` 2). New code reads v1 and v2 and writes v2.
+A v2 file keeps the `sessions` array and every v1 field of each entry, `pid` and `runnerPid`
+included, so the R guard (stack-release.sh:2094-2095) reads it unchanged (C1).
+A v1 entry has no `session`, and it is treated as the thread's legacy default session (D9). After a
+rollback, a 264a host refuses to start with exit 3, "sessions file is malformed"
+(session.mjs:241 via launcher.mjs:237), and the README's remedy applies: check which sessions still
+run, then move the file aside. Tradeoff: a rollback while runs are live needs a hand step, but it
+fails loudly at host start instead of leaving held claims. The rollback note names this refusal.
+
+## D9 consumer audit, threadsState and threadIdle (sage-2707 audit addition)
+
+These are trusted launcher APIs, not socket verbs (packages/bus/README.md:288, :294), and they're
+reached only through the host op table (packages/bus/src/process.mjs:23-24, :54-55;
+runtime.mjs:561-562). At aca16404c:
+
+| Consumer | file:line | What it reads |
+| --- | --- | --- |
+| launcher thread pass | packages/cli/src/launcher.mjs:596, :611 | every row's `thread`, `holder`, `pending`, `unread`; threadIdle `{business, thread, run}` |
+| bus end-launch.test | packages/bus/tests/end-launch.test.mjs:193, :198 | whole rows (deepEqual), and `not-holder` |
+| bus thread-migration.test | packages/bus/tests/thread-migration.test.mjs:139 | `thread, holder, pending, unread` |
+| bus threads.test | packages/bus/tests/threads.test.mjs:352, :375-376, :407, :416 | rows keyed by `thread` |
+| cli launcher.test stub | packages/cli/tests/launcher.test.mjs:989 | wraps `host.op`, counts the op name |
+| cli lifecycle.test stub | packages/cli/tests/lifecycle.test.mjs:90-91, :749-752 | passes through, wraps `runtime.threadsState` |
+| cli thread-launch.test stub | packages/cli/tests/thread-launch.test.mjs:90-91, :163, :264-265, :304, :348, :443 | `.find(x => x.thread === ...)` then `holder` |
+
+Console and webui: no reader. `git grep threadsState` over packages/webui, packages/control-board and
+scripts has no hit. The webui reads `thread.claims` (packages/webui/src/serve.mjs, cli.mjs,
+public/bus.js, tests shell.test, thread-claims.test, thread-claims-browser.test). This row doesn't change
+`thread.claims`. The webui's `session.launched` line reads only `harness` and `address`
+(public/bus.js:151), so a `session` field in the body changes nothing it shows.
+
+## D9 consumer audit, workspace and native store
+
+| Consumer | file:line at aca16404c | Change in this row |
+| --- | --- | --- |
+| `workspaceDir` | packages/seat/src/session.mjs:49 | unchanged, still the non-thread path and the legacy path |
+| launcher start, workspace | packages/cli/src/launcher.mjs:397-403 | a thread run uses its session workspace (Q1) |
+| launcher instance delete (V5) | launcher.mjs:841-866 | also retires the instance's session workspaces |
+| `threadStore` | session.mjs:59-74, launcher.mjs:388-393 | becomes the legacy branch of `sessionStore` |
+| sandbox binds | packages/seat/src/sandbox.mjs:102-125 | the workspace gets a separate in-sandbox path |
+| harness `transcript()` | packages/plugins/harness/pi/index.mjs:75, claude-code/index.mjs:130 | none: `{session, projects}` keeps its shape |
+| in-sandbox readers of `session.workspace` | packages/harness/src/runner.mjs:332, :386, :475-476; bundle.mjs:258, :327; gate.mjs:152 | none: they get the in-sandbox path |
+| task runner workspaces | scripts/mosaic-task.mjs:351 | none: `<dataRoot>/workspaces/<task>` is a legacy root the overlap check covers |
+| git socket | packages/cli/src/git-socket.mjs:7 | none: it takes bundles and reads no workspace |
+
+## Pi and Claude resume, read offline (sage-2707: no pi run while P19 383 holds)
+
+Pinned `@earendil-works/pi-coding-agent` 0.85.1 (package.json:8), read in node_modules/.../dist:
+
+- adapters/pi/adapter.sh:46-52 passes `--session-dir "$MOSAIC_SESSION_DIR"`, plus `-c` when that
+  directory isn't empty.
+- With `-c` and an explicit session dir, `continueRecent` filters by cwd
+  (core/session-manager.js:1244-1250). It opens the newest `*.jsonl` whose header cwd resolves to
+  the process cwd (:397-410, `sessionCwdMatches` :393-395). With no match, it makes a new session in
+  the same directory (:1251), and nothing fails.
+- Both adapters `cd "$MOSAIC_WORKSPACE"` (pi adapter.sh:34-36, claude adapter.sh:46-47), the runner
+  sets `MOSAIC_WORKSPACE` from `session.workspace` (runner.mjs:386), and the sandbox binds the
+  workspace at its host path (sandbox.mjs:112, `argsOf` :128-131). So a legacy transcript's header
+  cwd is `<dataRoot>/workspaces/<business>/<instance>`.
+- Opening a file whose header cwd doesn't exist exits 1 in a headless run (main.js:536-548,
+  core/session-cwd.js:2-15). So `--session <file>` is no way around it either.
+
+Result: a legacy default session run from `session-workspaces/.../<session>` as its cwd would
+quietly start a new pi conversation, and its old history wouldn't resume. Claude Code keys its
+projects directory on the cwd, and the adapter resumes by id (claude adapter.sh:53-59). I haven't
+read the Claude binary, so whether `--resume` finds an id saved under another cwd is unverified. I
+treat it as the same break.
+
+## Design
+
+### Legacy mapping (D9, M7)
+
+A session is legacy when its row has `op='legacy-3l'`, which the migration gave every pre-3l thread
+with `id` = the thread id (3k-3l.sql:50-51). The rule reads only an immutable row, never the disk.
+
+- A legacy session keeps the thread's existing store, `threads/<b>/<t>/{session,claude-projects}`
+  (`threadStore`, unchanged).
+- Its workspace is new (Q1), `session-workspaces/<b>/<instance>/<session>/`, but inside the sandbox
+  it's bound at the legacy path `workspaces/<b>/<instance>`. The cwd its transcripts recorded is
+  unchanged, so pi `-c` and Claude `--resume` find them. The host's old instance workspace isn't
+  mounted, so it stays as history (Q1, D9). Every in-sandbox reader gets the sandbox path (table
+  above), and host git (`ensureCheckouts`) gets the host path.
+- Every other session uses D2's layout for both, and its sandbox path is its host path.
+- A thread made by 264a's `thread.create` has a random default id (broker.mjs:5993), so this rule
+  doesn't call it legacy. A 264a launcher ran that thread from `threads/<b>/<t>/session`. A start
+  for a session whose thread has no legacy-3l session, while `threads/<b>/<t>/session` exists,
+  refuses `legacy-store-unmapped`, so it fails closed and doesn't start a fresh conversation. The
+  re-pin plan lists such threads. If 264a and 264b go live together, there are none.
+
+### Broker
+
+- `threadsState(business)`: one row per session of every thread, not just the default, ordered by
+  thread seq, then session seq. Each row adds `session` and `legacy` to the fields it has now.
+  `#threadRows` joins `sessions` in place of `session_defaults` (broker.mjs:868-896).
+- `threadIdle(record)`: same keys. The row is the one for the session the run was bound to
+  (`#runs` entry, broker.mjs:927), so a run is idled against its own session.
+- `bindLaunch`: no new refusal. It already takes `session` and puts it in the body (broker.mjs:945).
+  The broker half of the single-writer fence is the per-session claim that exists now:
+  a second claim on a held session refuses `thread-already-held` (broker.mjs:4815). A
+  `session-running` refusal at bind would break two old tests that bind a second unended run to one
+  session and then expect the claim refusal (threads.test.mjs:88-89, sessions.test.mjs:119-120).
+  The launcher's registry check and the workspace lock are the host halves.
+
+### Seat
+
+- `sessionStore(dataRoot, business, thread, session, { legacy })`: the legacy branch is
+  `threadStore`. Otherwise it's `threads/<b>/<t>/sessions/<s>/{session,claude-projects}`, with
+  `threadStore`'s checks on every segment and directory (ID_PATTERN, no `..`, owner, 0700, no
+  symlink), giving the same `{dir, session, projects}` shape.
+- `sessionWorkspace(dataRoot, business, instance, session, { live })`: makes
+  `session-workspaces/<b>/<i>/<s>` 0700 with the same checks, then compares realpaths. It refuses
+  `session-workspace-overlap` when the path is inside, or contains, `<dataRoot>/workspaces` (legacy,
+  `.retired` and task workspaces) or any workspace in `live`. A segment that fails the lstat checks
+  refuses `session-workspace-unsafe` first (C3).
+- Lock: `lockSessionWorkspace(path, owner)` makes `<path>.lock` with O_EXCL, next to the workspace
+  and outside every mount. It holds the host's pid and start time, and the run's outer pid and
+  start time once spawned. A lock whose host and run are both gone is stale and is taken over.
+  Anything else refuses `session-workspace-busy`. start() takes it before `ensureCheckouts` and
+  releases it only after the run's `close` (proven exit) or a failed start.
+- `Registry`: entries gain `session` and `workspace`, and the key becomes `running(business,
+  instance, thread, session)`. `SESSIONS_VERSION` is 2, and `readSessions` reads 1 and 2.
+
+### Launcher
+
+- check(instance, { thread, session }): `instance-running` is per session. Q2 adds
+  `legacy-run-live`: a start for a thread instance refuses while any registry entry or unrecovered
+  leftover for that instance has no `session`.
+  A survivor is still bound and ended host-lost (launcher.mjs:1030-1040), so the broker sees its
+  claim free while it runs. A released claim doesn't prove the run has exited. Only this launcher
+  check holds the instance then, for sessionless survivors and thread starts only (duckworth-0158
+  N1). recover() waits up to 5 s per leftover for the outer pid and the runner (N2, see "recover()
+  change, for a design read").
+  Residual (duckworth-0159 N3): a survivor with a `session` counts nowhere, because :309 counts
+  only sessionless entries. The same session could start again while the old runner lives. The
+  lock doesn't catch it either: it records the outer pid as runPid (launcher.mjs:574), so with the
+  outer pid gone it reads stale and is taken over. The base has the same exposure. In practice
+  the runner dies with its pid namespace when the outer bwrap is killed.
+- start(): the store and workspace come from the session. The lock covers checkouts and spawn.
+  `bindLaunch` gets `session`, and the entry records it.
+- threadPass(): one pass per session row. `startedFor` and `refusedThreads` key on the session.
+  The `thread-idle`, `refused` and `end` records add `session`.
+- recover(): passes the entry's `session` when present.
+- remove(): moves `session-workspaces/<b>/<instance>` to `session-workspaces/<b>/.retired/<instance>-<transaction>`
+  after the legacy move, and records `workspace-retired` with that path.
+- Admission (RULINGS Q5, D8): a named session starts only through threadPass, then check(), then
+  start(), so it passes through every `admission()` call start() makes now (launcher.mjs:382-524).
+  There is no second start path.
+
+### Sandbox
+
+`binds()` takes `workspaceDest` (default `workspace`), the in-sandbox path for the rw workspace
+bind. The store binds keep their shape: only the selected session's two directories.
+
+## Tests
+
+T1 run half, with the fake harness adapter from thread-launch.test (no pi or Claude binary):
+two sessions of one thread run at once, each with its own sentinel history, store and workspace;
+stopping one and restarting it resumes only that one. The legacy session resumes from the old
+store at the old in-sandbox cwd. A non-thread instance gets `thread-not-allowed`.
+
+T2: a second start on a live session's workspace refuses `session-workspace-busy` before any
+checkout; a hand-made live lock from another host refuses; a symlinked or nested session root
+refuses `session-workspace-unsafe`; a workspace root symlinked into the session path refuses
+`session-workspace-overlap`; a store segment symlink refuses `thread-store-refused`. Under a halt, a
+named session's start refuses `processing-halted` and spawns nothing. A generation change after
+check() refuses `processing-generation-changed` and spawns nothing (M6).
+
+N1 (DEFERRED.md:474-480): `refusedOnce` in packages/bus/tests/sessions.test.mjs pins `actor_run`.
+
+## Old assertions that change (C4, duckworth-0150, mos-1285)
+
+Every pre-existing test that reads threadsState at aca16404c, checked against one row per session:
+- packages/bus/tests/end-launch.test.mjs:193-195 deepEquals the whole row. This one changes. Mos
+  allowed it in mos-1285 as an exact update: still a whole deepEqual on threadsState('demo'), still
+  exactly one th-1 row, every old key and value the same, and two new keys with literal values,
+  `session: 'th-1'` (the fixture's legacy default, :14-15) and `legacy: true` (op `legacy-3l`). The
+  commit message names mos-1285 and the old row shape.
+- packages/bus/tests/threads.test.mjs:352, :375-376: fromEntries keyed by thread. Neither test, and
+  neither fixture (`twoThreads`), makes a second session (no `session.create` in the file), so each
+  thread still has one row. Unchanged.
+- packages/bus/tests/thread-migration.test.mjs:139: maps four fields of a one-session thread.
+  Unchanged.
+- packages/cli/tests/thread-launch.test.mjs:163, :304, :348, :443: `find` by thread. One session
+  per thread in every fixture. Unchanged. The default session also sorts first, by session seq.
+- packages/cli/tests/lifecycle.test.mjs:90-91, :749-752 and thread-launch.test.mjs:90-91, :264-265
+  pass threadsState through. Unchanged.
+- packages/cli/tests/thread-launch.test.mjs:182 (found in the code round, after mos-1285). The test
+  makes threads with `thread.create` (:171-172), so A's default session has a random id and op
+  `default` (broker.mjs:5992-6009), not `legacy-3l`. The brief puts every new store in D2's layout
+  (t1-sessions.md:100, :294), so A's store moves to `threads/acme/<A>/sessions/<A's session>`. The
+  proposed exact update changes only the `store` path on :182, taking the session id from
+  `threadsState` (the test already reads it at :162). Lines :183-185 stay as they are. One added
+  line asserts that `threads/acme/<A>/session` doesn't exist. This goes to Mos through Sage, and no
+  code commit edits the file until Mos rules.
+- The same file's :375 (`../evil` refuses `thread-store-refused`) and :381, :384 (no `threads`, no
+  `workspaces`) stay unchanged: start() checks the store before it makes the session workspace.
+
+Any other old assertion the code changes goes to Mos through Sage before the commit. After the
+first run I survey every touched file whole for one.
+
+## Mutants
+
+Each kill counts only as an assertion failure in a test that runs that path, never a throw or
+another fence (sage-2707, C3 duckworth-0128).
+
+| Id | Change | Killed by |
+| --- | --- | --- |
+| M1 | `sessionWorkspace` drops the session segment: a thread's sessions share one checkout | T2 same-thread fence: `session-workspace-busy`, no second spawn |
+| M2 | the session segment becomes the session name: two threads' `default` sessions share one checkout | T2 cross-thread fence, same assertion |
+| M3 | the lock ignores a live holder | T2 hand-made lock: `session-workspace-busy` |
+| M4 | the realpath overlap check removed | T2 nesting: `session-workspace-unsafe` |
+| M5 | the launcher passes the thread's default store for every session | T1: the second session's sentinel history is its own |
+| M6 | start() calls a no-op in place of every `admission(admissionGeneration)` (launcher.mjs:382, :396, :407, :420, :495, :499, :520, :524) for a non-legacy session | the generation test (below): `assert.rejects` with code `processing-generation-changed`, then `assert.equal` spawn count 0 |
+| M7 | the legacy session maps to `sessions/<s>/` | the D9 test: the legacy run resumes the sentinel in `threads/<b>/<t>/session` |
+| M8 | a non-default session binds its workspace at the legacy `workspaces/<b>/<instance>` in-sandbox path | T1: `assert.equal` that a named session's in-sandbox workspace is its `session-workspaces/<b>/<instance>/<s>` path, read from the fake adapter's recorded cwd (sage-2717) |
+| M9 | `#threadRows` back to the default session only | T1: the named session's message starts its run |
+| M10 | the `legacy-run-live` refusal removed from check() | the legacy-run-live test: `outcome` equals `[[S, "legacy-run-live"]]` |
+| M11 | recover() keeps a survivor only when the outer pid outlived the kill (the runner ignored) | the legacy-run-live test, same assertion |
+| M12 | check() reads every survivor, live or not | the legacy-run-live test: after the runner exits, `settle` gives one launch |
+
+M6 and the halt fences (C2, duckworth-0150). A plain halt never reaches start(). tick() admits first
+(:579) and runs no thread pass while halted (:583), and check() admits at :282 and :376. The kill
+test lands the change between check() returning and the spawn:
+- The fake host's `credentialStatus` op (start() calls it at :395, after check() returned and after
+  :382) records that it ran, then halts and resumes processing through `transition`. That leaves
+  `halted` false and a new generation (state.mjs:115-121).
+- The tick and check fences see no halt, and start()'s second check() (:419) takes its own
+  generation, so it passes too. withProcessing at :406 has no parent context (the launcher never
+  opens one) and checks only `halted`, so it passes. The one fence left is start()'s own
+  `admission(admissionGeneration)`, which refuses `processing-generation-changed` at :396.
+- The test asserts the hook ran (check passed), then the refusal code, then no spawn.
+M6 removes all eight calls because removing only :382 leaves :396 to refuse. The T2 halt test stays
+as written in Tests, but it doesn't kill M6 and isn't counted for it.
+
+M4 and the segment checks (C3, duckworth-0150). The lstat check on each segment refuses a symlinked
+session root before the realpath compare runs, so the compare gets its own code,
+`session-workspace-overlap`, and the segment check keeps `session-workspace-unsafe`. M4's kill input
+passes every segment check: `<dataRoot>/workspaces/<b>` is a symlink to
+`session-workspaces/<b>/<i>/<s>`. Each segment of the session path is a real directory, and only the
+realpath compare finds that the session path contains the legacy workspace root. The test asserts
+`assert.rejects` with code `session-workspace-overlap` and no spawn. With M4 the start spawns, so
+the rejection assertion fails. The segment test keeps asserting `session-workspace-unsafe` for a
+symlinked session root, so it tells the two codes apart.
+
+Duckworth's check for the early look (sage-2717): a non-default session never gets the legacy
+`workspaces/<b>/<instance>` bind path. Only a session with op `legacy-3l` takes `workspaceDest`
+set to the legacy path. M8 is that mutant.
+
+### Code round corrections (local, not yet run)
+
+Written before the first run, which waits for Sage's CLEAR.
+
+- Tests live in two new files: packages/cli/tests/session-launch.test.mjs (T1, D9, T2) with a new
+  stub adapter, fixtures/session-adapter.mjs (thread-adapter.mjs's turn plus the run's cwd and its
+  entries; thread-adapter.mjs is unchanged), and packages/seat/tests/session-workspace.test.mjs
+  (sessionStore, sessionWorkspace, the lock, sessions.json v2). The new launch tests copy
+  thread-launch.test.mjs's host helpers rather than change them, so no other old test file changes.
+- M1 and M2 kill in T1, not in a busy test. A test that expects `session-workspace-busy` still
+  passes when sessions share a path. T1 starts thread A's default and named sessions and thread B's
+  default at once and asserts no refusal, three live entries and three distinct cwds. With M1 (A's
+  two sessions) or M2 (the two `default` sessions) the second start refuses busy, and the
+  `deepEqual(refused, [])` assertion fails.
+- Every wait a mutant could stall returns what it last saw (`settle`, `outcome`), so a kill is the
+  assertion after it, never a timeout. That covers M3, M4, M6 and M9, where the mutant's run starts
+  or never starts.
+- The M6 test also carries the halt fence. start() admits right after credentialStatus, before it
+  makes the session workspace, so the first refusal leaves no workspace. On the retry the recheck's
+  mayLaunch op halts after the lock is taken. The recheck refuses `processing-halted`, and the test
+  asserts that the lock is gone (startLocked's throw releases it).
+- The lock's release() runs once. A second call would unlink a lock another host has written since.
+- `legacy-run-live` now has a test (Duckworth's early warning in duckworth-0156). Before, recover()
+  kept a survivor only when a bwrap pid outlived SIGKILL, which a test can't make. It also ended
+  a leftover whose runner still ran once the outer pid was gone, so the refusal never saw that run.
+  Now recover() waits up to 5 s on the outer pid and the runner. It keeps any leftover where
+  either still runs, and logs `run <run> (<instance>) from an earlier host still runs`. check()
+  reads only survivors that still run, so a start goes ahead once the run has exited. The test
+  writes a sessionless leftover whose runner is a sleep child and whose outer pid is gone. It
+  asserts the refusal, no launch, no session workspace, and then one launch after the child is
+  killed. M10-M12 cover it. The registry and pre-recover leftover terms stay. A launching host
+  can't reach them for a thread instance (check() refuses `session-required` first, and listen()
+  runs recover() first), but they cost nothing.
+
+### recover() change, for a design read (sage-2750)
+
+Sage asked for this because recover() predates the row. Lines are launcher.mjs at 3d29876df plus
+the log-line edit after it, against aca16404c:947-976.
+
+- What stays the same: every leftover is still bound and ended as host-lost (:1030-1040, base
+  :963-973). The survivor push (:1026-1029) comes before the bind, and the loop carries on past it,
+  so a kept leftover isn't left unended. Its bus run ends and its role is released, as in the base.
+  The kill is unchanged too: SIGKILL only to a live outer pid whose cmdline is bwrap (:1018-1021,
+  base :956-959).
+- What changes:
+  1. The wait after the kill runs while the outer pid or the runner still runs (:1023). In the
+     base it waited on the outer pid only (base :961).
+  2. A leftover where either still runs after that goes on `survivors` and is logged (:1026-1029).
+  3. check() counts a survivor only while it still runs (:309, `liveRun` at :255).
+- a. Nothing in this row kills a kept leftover's runner. The base didn't either: it ended the run
+  and let the runner go on, unseen. The run is still ended (:1038), so the runner holds no role.
+  Its capability died with the old broker (:1004-1008). The only thing a survivor does is refuse
+  `legacy-run-live` for sessions of its instance while it lives, and each refusal is a `refused`
+  launch record. For an operator, the log line names the run, the outer pid and the runner pid
+  (:1028). Killing that pid clears it. check() re-reads liveness on each pass (:309), so the next
+  tick starts the session with no host restart. If bindLaunch refuses (:1032-1036), the run stays
+  unended and logged, as in the base (base :965-969). The survivor entry doesn't change that.
+- b. The 5 s wait only applies inside the kill branch (:1018-1024), so only to a leftover whose
+  outer bwrap was still alive. A leftover with nothing running doesn't wait, and nor does one
+  whose outer pid is gone: neither enters the branch. The wait ends as soon as both processes are
+  gone. The base already waited up to 5 s in the same branch. It can now last longer only if the
+  runner outlives its outer pid. With `--unshare-pid` and `--die-with-parent` the runner dies when
+  the namespace goes.
+- c. Old tests that run recover() with a launching host and a leftover:
+  - launcher.test.mjs:885-943, the "host that died hard" pair. It asserts the outer pid and the
+    runner are gone right after setup (:917-918), the host-lost log (:919), one session.ended
+    (:920-922), one end record (:924), an empty sessions.json (:925), and then "already ended" for
+    a leftover whose processes are dead (:937-941). None of these expected values change.
+    Waiting on the runner makes :918 hold sooner, not later. The new log line can appear only if
+    a process outlives 5 s, which would already fail :917-918. The dead-process third host takes
+    no wait and logs no survivor.
+  - launcher.test.mjs:950-971 (`--no-launch`) returns before the loop (:1012-1015). It's
+    unchanged.
+  - stack-release.test.mjs:1928 writes `{pid: process.pid}` with no start times. `liveRun` is
+    false for it, and that test reads it through bus status, not recover().
+  - verbs.test.mjs:213 and the seat tests build a Registry with no launcher, so they aren't
+    affected.
+
+  No expected value in an old test changes, so nothing goes to Mos under mos-1272 for this. If a
+  run shows otherwise, I stop and send it before any change.
+- Fallback, if Duckworth or Sage doesn't accept this: revert recover() and check() to the base
+  behaviour, keeping only the sessionless check over the registry, the pre-recover leftovers and
+  bwrap survivors. Drop the legacy-run-live test and M10-M12, and name the untestable path here.
+- N1 (refusedOnce pins `actor_run`) changes a pre-existing test. Mos allowed it in mos-1292 (copy
+  mos-1294) on mos-1285's terms, and it's committed as a81b3446e, with a closing line on the
+  DEFERRED.md entry.
+
+## Rollback
+
+264b writes sessions.json v2. A 264a host given that file refuses to start, exit 3, "sessions file
+is malformed" (session.mjs:241 via launcher.mjs:235-237). Session workspaces and per-session stores
+stay on disk, and a 264a host never reads them.
+
+What else refuses on a 264a stack while a v2 file is in place (C1, duckworth-0150). All of them go
+through the 264a readSessions, which throws SeatError "sessions file is malformed" (session.mjs:241):
+- `mosaic launches list` and `mosaic stop <run>`: exit 2 with that message (cli.mjs:526, :465).
+- the bus `limits` verb: `settings.read()` throws (runtime.mjs:149, :292, launch-settings.mjs:166),
+  so the verb fails with that message. The limits the bus enforces still work, since `current()`
+  doesn't read the file. I haven't traced the code a bus caller sees for that throw.
+- the Console settings view: launch shows `not-configured` (reads.mjs:234).
+- the R guard keeps working (above).
+All of these read the file on each call, so step 3 below clears them without a restart.
+
+Console restart: the shape of what the Console reads doesn't change. usageCounts still counts
+entries per provider and model, and reads.mjs keeps its fields. What changes is the file, and a
+Console still on 264a code shows launch `not-configured` from the first v2 write until a human
+restarts it on a 264b copy. The re-pin that carries this row needs its Console restart (Jason or
+Mos), done right after the host's, as it does for a read-shape change. Step 3 covers the reverse
+case after a rollback with no restart.
+
+Seeing live sessions without `launches list` (C1): read the file directly the way the R guard does.
+For each entry in `sessions`, a session is live if `/proc/<pid>/stat` or `/proc/<runnerPid>/stat`
+exists and its state isn't `Z` (stack-release.sh:2080, :2095). For a stronger check, compare
+`runnerStartTime` with field 22 of `/proc/<runnerPid>/stat`, which is what readSessions' `live`
+does through startTimeOf.
+
+How a rollback host gets past exit 3 (sage-2717), run on NEW before the rollback activate:
+1. Stop the bus host on NEW. Its stop SIGKILLs what is left and ends those launches as `killed`
+   (packages/cli/README.md:470).
+2. Read `sessions.json` with NEW's readSessions (`mosaic launches list`, cli.mjs:526-530). Go on only
+   if it shows no live entry: every listed runner is gone (pid and start time no longer match). If
+   NEW's CLI is already gone, use the /proc read above.
+3. Move `sessions.json` aside under a dated name, never delete it (packages/cli/README.md:478).
+4. Then roll back. The 264a host starts with no leftover file and nothing to recover.
+A live entry at step 2 stops the rollback until that run ends or `mosaic stop <run>` ends it.
+
+Which re-pin plan step carries it: no re-pin plan names 264b yet. The 84f53540d re-pin carries 264a
+only. The first re-pin whose NEW includes this row puts steps 1-4 in its "Stop conditions and
+rollback" section, before `rollback --plan`, the section the 10-24 plan uses
+(agents/sage/work/window-plan/repin-1024-draft1.md:144-148). I'll name it to whoever drafts that
+plan when the row lands.
+
+## Run limits (sage-2707)
+
+Each node --test file and suite runs under `timeout -k 10 LIMIT`, and each scripts/test-*.sh under at
+least 3x its longest recorded run. Records are from agents/gizmoduck/work/queue-264/BUILD.md "Run
+limits" unless named.
+
+| Run | Longest recorded | LIMIT |
+| --- | --- | --- |
+| stack-release.test alone | 575 s (264a r12 survey) | 1800 s |
+| each other touched file alone | 15 s (thread-launch.test); the rest have none | 300 s |
+| bus suite | 23 s | 900 s |
+| cli suite | 591 s (row 281 gate) | 1800 s |
+| webui suite | 57 s | 600 s |
+| seat suite | none recorded | 600 s |
+| harness suite | none recorded | 600 s |
+| each scripts/test-*.sh | 42 s (test-foundation) | 900 s |
+| each mutant probe | 1 s wall (264a M1-M4) | 300 s |

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

@@ -0,0 +1,156 @@
+# Row 270 build notes (Gizmoduck)
+
+264c, T1 item 7: the compaction count, exact for Pi. Gitea #1775. Brief:
+docs/plans/2026-10-21_t1-sessions.md section 4 (D6), with RULINGS Q6 (Pi exact, Claude `unknown` with
+the Vikunja 466 reason). Reviewer Duckworth, gate owner Sage. Branch gizmoduck/row270 from
+gizmoduck/row269 2d0d407ba (sage-2797, Q220). Row 269 lands first, and this branch rebases with it.
+
+## Contract for 264e (pinned for Huey, row 272)
+
+The surface reads the count from `session.list`. Each row gains one key, `compactions`:
+
+```json
+{ "id": "...", "at": "...", "name": "...", "predecessor": null, "default": true,
+  "compactions": { "count": 3, "known": true, "reason": null, "at": "2026-10-22T16:40:00.000Z" } }
+```
+
+- `known: true`: `count` is the exact number of completed compactions over the session's lifetime,
+  an integer of 0 or more. `reason` is null.
+- `known: false`: `count` is a lower bound (an integer) or null when there is no bound. Show
+  `unknown` with the reason, never a bare 0. `reason` is one of:
+  - `harness-unsupported`: Claude Code, until Vikunja 466 pins its format.
+  - `not-collected`: no collection has run for this session yet.
+  - `store-unreadable`: a store file couldn't be read, or isn't a Pi session.
+  - `store-incomplete`: a stream file got shorter, was replaced, or has unreadable lines, so
+    entries may be missing.
+- `at`: ISO time of the last collection, or null if none has run.
+- A clear successor (264d) starts at `known: true, count: 0` once its first collection runs. The
+  retired session keeps its row and its count.
+- On a base without 264c, the key is absent. The surface shows the count as unavailable, which is
+  what Huey's plan does (Sage Q1).
+
+No new verb and no new input key. Access is `session.list`'s: the thread owner, or the tester.
+
+## Pi source contract (pinned offline at 0.85.1)
+
+`@earendil-works/pi-coding-agent` 0.85.1 (package.json pin), read in node_modules:
+
+- One line per completed compaction: `{"type":"compaction","id":...,"parentId":...,"timestamp":...,
+  "summary":...,"firstKeptEntryId":...,"tokensBefore":...}` from `appendCompaction`
+  (dist/core/session-manager.js:818-833). The two callers write it after the summary exists
+  (dist/core/agent-session.js:1539, :1826). A start or an aborted compaction writes no entry.
+- What doesn't count: a `message` entry with role `compactionSummary` (a rendered summary,
+  docs/session-format.md:154), any message inside a compaction's `retainedTail`, and
+  `branch_summary` entries.
+- Identity: a compaction is its (entry `id`, entry `timestamp`) across the whole store, not per
+  stream. Pi's ids are unique only within a file (`generateId(this.byId)`, session-manager.js:21-29),
+  and `createBranchedSession` (:1093-1140) and `forkFrom` (:1264-1300) copy entries into a new
+  `*.jsonl` in the same directory with their id and timestamp unchanged. A (stream, id) key would
+  count a compaction made before a branch or fork once per copy. Reported to Sage in
+  gizmoduck-0262; accepted in sage-2814. The ledger line records the stream it was first seen in.
+- Residual (sage-2814): two distinct compactions that share an id and a timestamp count once. That
+  needs an 8-hex collision between two files at the same millisecond. It is the one undercount
+  this key allows, and it never overcounts.
+- Partial trailing line: wait. Only lines that end with a newline are read, as
+  packages/conversation/src/pi.mjs:35-38 does.
+
+## Storage: A, a host ledger (sage-2806)
+
+Sage ruled A over B. A keeps the count in `<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.

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

@@ -0,0 +1,417 @@
+# Row 271 build notes (Gizmoduck)
+
+264d, T1 item 7: clear context through row 194's lifecycle stop. Gitea #1776. Brief:
+docs/plans/2026-10-21_t1-sessions.md section 5 (D7, D8), with RULINGS Q4 (rebase onto 194's
+landing, no 194 code copied). Reviewer Duckworth, gate owner Sage. Built after row 270 (sage-2797).
+
+## Contract for 264e (pinned for Huey, row 272)
+
+Proposed, pending Duckworth's review. Any change goes to Huey before it's pushed.
+
+### session.clear
+
+Input, every key required, with no other key (`keys()` as `session.create` does):
+
+```json
+{ "thread": "<thread id>", "session": "<session id>", "revision": 0, "confirm": true, "op": "<op key>" }
+```
+
+- `revision`: the integer from the session's `session.list` row (below). It names the state the
+  human confirmed.
+- `confirm`: the literal `true`. Anything else refuses `confirm-required`.
+- `op`: the caller's operation key, an identifier, unique per thread. It is also the idempotency
+  key.
+
+Answer, the operation envelope:
+
+```json
+{ "op": "...", "thread": "...", "session": "<old id>", "successor": "<new id>",
+  "state": "clearing", "start": null, "refusal": null, "at": "..." }
+```
+
+- `successor` is bound when the request is accepted, before any asynchronous work. It is a
+  lowercase UUID, `xxxxxxxx-xxxx-8xxx-Vxxx-xxxxxxxxxxxx` with V one of `8`, `9`, `a`, `b` (RFC 9562
+  version 8, variant 10). It's the first 32 hex digits of sha256(business, thread, old session, op)
+  with those two nibbles set. It matches the UUID test every session selector uses
+  (cli/src/thread.mjs:41), so 264a's selectors and deep links take it unchanged (Huey's question,
+  gizmoduck-0274).
+- `state`:
+  - `clearing`: accepted. Sends and launches to the old session are fenced, and the drain or stop
+    is in progress.
+  - `uncertain`: stop proof isn't available. Capacity stays reserved and nothing is forced. It
+    resumes when proof arrives.
+  - `retired`: the old session is retired and the successor exists, in one bus transaction.
+  - `failed`: no retirement happened, and `refusal` names why. The old session stays as it was.
+  - `cancelled`: a human cancelled the clear (session.cancel below, sage-2828). `retired` is
+    null if it was cancelled before retirement and true if after. `start` stays as it was.
+- `start` is null until `state` is `retired`. After that it is `held` (waiting for capacity),
+  `started`, or `refused` with `refusal` set.
+- `at`: the time of the latest state change.
+
+Repeats and conflicts:
+
+- The same `op` with the same payload returns the current envelope and starts nothing new.
+- The same `op` with any other payload refuses `session-op-conflict`.
+- A new `op` on a session with an open clear refuses `session-clearing`.
+- A `revision` that isn't current refuses `session-revision-stale`.
+- Other refusals: `session-not-found`, `session-retired`, `processing-halted` (D8: a halt refuses
+  and changes nothing), and the owner refusals `session.create` already gives.
+
+### session.cancel (sage-2828)
+
+Input, every key required, with no other key:
+
+```json
+{ "thread": "<thread id>", "op": "<the clear's op>", "confirm": true }
+```
+
+This is a human write with the same owner or write-grant check as session.clear, rechecked under
+the session fence. An agent caller refuses as session.clear does. `confirm` must be the literal
+`true`, or it refuses `confirm-required`. The answer is the operation envelope with `state`
+`cancelled`.
+
+- Before retirement, cancel stops the replacement:
+  - The successor id is never created or started.
+  - Messages already cancelled stay cancelled and are never replayed.
+  - The old session is `active` again.
+- After retirement, cancel stops the startup only:
+  - No further start attempts are made.
+  - Both sessions stay as they are. Nothing un-retires.
+- Repeats and refusals:
+  - A repeat with the same payload returns the original cancelled envelope.
+  - The payload digest covers `{thread, op}` and the calling principal. A cancel from a
+    different principal, or a payload that otherwise differs, refuses `session-op-conflict`.
+  - An op that isn't a clear refuses `operation-not-found`.
+  - A cancel of a clear whose successor has started refuses `session-clear-complete`.
+
+### session.operation
+
+Input `{ "thread": "...", "op": "..." }`. Answer: the same envelope. Unknown op refuses
+`operation-not-found`. A CLI that waits 30 s and then reports pending uses this to poll. A retry
+queries the op here and never resends the clear.
+
+### session.list additions
+
+Each row gains:
+
+- `state`: `active`, `clearing`, `uncertain` or `retired`. A cancelled clear leaves the session
+  `active` if cancelled before retirement, and `retired` if after.
+- `revision`: an integer. 0 at creation, plus 1 for each lifecycle event on that session.
+- `successor`: the successor's id once retired, otherwise null.
+
+A successor row has `predecessor` set (264a's column). When the cleared session was the thread's
+default, `default` moves to the successor at retirement. Clearing any other session leaves the
+default where it is.
+
+On a base without 264d, `session.clear`, `session.cancel` and `session.operation` refuse
+`unknown-verb`, and the rows have no `state`,
+`revision` or `successor`.
+
+## Design (draft 2: Sage's sage-2826 rulings and Mos's mos-1315 folded in)
+
+Measured on 2d0d407ba, which already holds row 194 (ab7377bcf). Nothing has been run. Draft 1
+was 3105de889. Draft 2 changes where the drained acknowledgement comes from (stage 2): threadIdle
+alone can't show that no child is running.
+
+### What exists and what doesn't
+
+- No per-session lifecycle state. `sessions` is append-only (schema.sql:370-372). The nearest
+  per-session state is the latest `thread_claims` row (op and generation). Row 194's lifecycle
+  is per instance, as `action.allowed` events keyed by a random request id (broker.mjs:622-666).
+- No cancelled state for an undelivered message. `deliveries.op` is a CHECK list on an
+  append-only table.
+- The runner calls `message.receive` only when no harness child exists:
+  - `runTurn`'s close handler SIGKILLs the turn's whole process group and clears `child`
+    (harness/src/runner.mjs:363-366).
+  - The loop asks receive only after the landing turns (:493-517) and after each message's
+    RESULT and `message.read` (:524-580).
+  - So a receive from the run is a completed turn boundary, made by the runner itself.
+- `threadIdle` (broker.mjs:865-877) is not enough. A landing turn reads no message, so `unread`
+  is 0 while its child runs. That is why landRelease keeps a run until it has asked receive
+  (`#received`, broker.mjs:363-365, :3815-3829).
+- There is precedent for acting at receive. `#limitStop` (broker.mjs:4147-4148, :4499-4520)
+  holds a stop until the run's next `message.receive` or `state`, so the turn's RESULT and read
+  land first.
+- The runner's SIGTERM handler escalates to KILL after 5 s only when `child` is set
+  (runner.mjs:405-423). With no child it sets `stopping`, the loop ends, and it exits 0.
+- The launcher's idle stop (launcher.mjs:683-701) sends a SIGTERM after a start-time check. The
+  exit proof is the run's `close` handler, which runs endRun, removes the registry entry and
+  releases the workspace lock (:600-641).
+
+### Clear, stage by stage
+
+1. **Accept** (`session.clear`, the human's write). One transaction:
+   - Check owner and grant (the `session.create` path), the revision, and that processing isn't
+     halted.
+   - Bind the successor before any async work:
+     - The id is deterministic, from a sha256 of (business, thread, old session, op), shaped as a
+       lowercase version 8 UUID (the contract above). 264a's ids are UUIDs, and every selector
+       checks the UUID shape, so draft 2's `ID_PATTERN` shape is replaced.
+     - The name is `<old name>-N`, with N the next free number in that thread. If a user already
+       made a session called `<old>-N`, the next free N is taken.
+     - Both go on the `requested` row, so a repeat of the op returns the same id and name.
+   - Append journal stage `requested`, with a payload digest for the repeat and conflict checks.
+   - The answer is `clearing`. From here:
+     - `thread.send` and `thread.open` naming the session refuse `session-clearing`. So does a
+       thread-only send whose default is that session.
+     - `threadsState` reports the session as `clearing`, so `threadPass` never starts it.
+     - Undelivered messages to it are cancelled with a receipt each (`message_cancels`), in the
+       same transaction. Old messages, results and run evidence don't change.
+2. **Drain, acknowledged by the runner.** Once the session is clearing, the run's
+   `message.receive` (the runner's call, at a turn boundary with no child) takes one transaction:
+   - It appends `drained` with the run and its claim generation, once.
+   - It delivers nothing and answers `[]`.
+   - The claim stays held. That holds the dequeue fence, so no turn and no child can start
+     before the stop, and the old capacity stays reserved.
+   Later receives keep answering `[]`. `state` needs no fence, because the runner asks it only
+   for a message it received. A clear on a session with no holder appends `drained` with run
+   null at accept.
+3. **Stop and proof** (launcher). For a `drained` op whose run and generation are still the
+   holder's, the launcher sends the idle stop's start-time-checked SIGTERM, the same reviewed
+   path. The runner has no child, so its handler never reaches KILL.
+   - The exit proof is the run's own `close` handler (endRun, registry remove, lock release),
+     recorded with `sessionClearExited {business, op, run}`. A signal, a revoked claim or a dead
+     socket is not proof.
+   - Under the host, the runner is pid 1 of its own pid namespace (`--unshare-pid --as-pid-1`),
+     so its exit ends anything a turn left outside its process group. **This is a build check**,
+     tested against bwrap, not assumed.
+   - With no run, the proof is that the launcher takes `lockSessionWorkspace` for the session and
+     the registry has no run for it.
+   - If no `drained` arrives (a turn still running, or a stuck child), the session stays
+     `clearing`. Capacity stays reserved, the successor stays unstarted, the status stays
+     actionable, and no signal is sent.
+   - Thirty seconds after the SIGTERM with no proof, `uncertain` is appended once. The launcher
+     keeps waiting and sends no second signal. A launcher restart finds the run among its row 269
+     survivors and waits on its identity.
+4. **Retire** (trusted op `sessionClearRetire {business, op}`), one transaction:
+   - Check the exit proof (the run is in `#ended`), or that there was no run.
+   - Append `retired`, and end the old claim.
+   - Insert the successor `sessions` row with the bound id and name, `predecessor` set to the old
+     id and `op` set to the clear op.
+   - Add a `session_defaults` row only if the old session was the default.
+   - Add a `thread_opens` row for the successor.
+   The old workspace stays at its path with no new writer mount. The successor gets its own store,
+   HOME and workspace from row 269's per-session layout, and a fresh checkout of the approved
+   revision. Nothing is copied.
+5. **Start** (D8). `threadPass` starts the successor from its `thread_opens` row through the same
+   `check()`/`start()` as any session. The launcher reports the outcome with `sessionClearStart
+   {business, op, start, refusal}`, which appends `start-held`, `start-started` or
+   `start-refused`, each with the next report counter. A held start keeps its op and spends no
+   work attempt.
+
+`failed`: a refusal at retire (for example, the owner lost the grant) appends `failed` with the
+code. Nothing is retired, and the old session goes back to `active` with its messages still
+cancelled.
+
+**Cancel** (sage-2828, re-ruling sage-2826 Q3, D7:263-264). `session.cancel` checks the owner or
+write grant under the fence and appends `cancelled`, attempt 0, in one transaction:
+- Before `retired` it releases the fence:
+  - If no `drained` was recorded, the drain is still waiting. Nothing is signalled, the clear
+    ends, and the run's next receive delivers again.
+  - If `drained` was recorded with the claim still held and no SIGTERM sent, the launcher skips
+    the stop because the op is no longer open. The run keeps its claim, and its next receive
+    delivers again.
+  - If the SIGTERM was already sent or the run has exited, the old session is `active` with no
+    runner. The next send launches it through D8 like any idle session.
+  - The successor row is never inserted, and the cancelled messages stay cancelled.
+  - The launcher reads the op state right before its SIGTERM, in the same tick as its start-time
+    check. A cancel after that point still finds a stopping runner, which takes the "active with
+    no runner" path above. This is a residual: a cancel can't take back a SIGTERM already sent.
+- A pre-retire cancel doesn't touch the claim (duckworth-0171 C2):
+  - A run that still holds it keeps it, generation and all. Releasing the fence only stops its
+    receives answering `[]`.
+  - A run already stopped ended its own claim through endRun, so there's nothing to change.
+  - Messages already cancelled keep their `message_cancels` rows and are never delivered.
+- After `retired` and before `start-started`, no further start attempts are made.
+  - The successor row and the default move stay. Nothing un-retires the old session.
+  - threadPass doesn't start the successor from the clear's `thread_opens` row: a cancelled op
+    suppresses that open.
+  - **My reading, stated for Sage:** the successor is then an ordinary idle session. A later new
+    send to it, including a thread-only send while it's the default, launches it through D8 like
+    any idle session. That's new input, not the clear's startup. The other reading leaves the
+    thread's default pointing at a session that can never start, because there is no verb to
+    move a default (the session verbs are list and create).
+- The race with the drained receive: `drained` and `cancelled` are each one synchronous
+  transaction on the open op. Whichever commits second sees the first. A receive after the
+  cancel finds no open op and delivers normally. A cancel after `drained` takes the "drained, no
+  SIGTERM yet" path above.
+- After `start-started` it refuses `session-clear-complete`.
+
+### Early look d2 folds (duckworth-0170, 98dc69f0c)
+
+- **B1, messages routed by `#sessionFor` (broker.mjs:722-748).** A role reply chain or a spec
+  notice can resolve to a clearing or retired session with no state check. Rule: **cancel with a
+  receipt at arrival.** The message is inserted (old results stay as evidence, D7 step 4) and its
+  `message_cancels` row goes in the same transaction, so it is never delivered and never pending.
+  The send's answer gains `cancelled: "session-clearing"` or `"session-retired"`, an additive key.
+  - Refusing was the other option. It would lose a teammate's RESULT outright: the runner only
+    logs a failed reply (harness/src/runner.mjs:575-580).
+  - Routing to the successor would replay old input, which D7 forbids.
+  - A spec notice resolves through `#defaultSession`. During a clear of the default session that
+    is still the old session, so the notice is cancelled. After retirement the default is the
+    successor, and a new notice is new input that goes there.
+  - `threadsState` reports `retired`, and `threadPass` starts neither a `clearing` nor a
+    `retired` session.
+- **B2, a claim racing a no-holder clear.** role.claim's thread path (broker.mjs:4827-4845)
+  refuses `session-clearing` for a session with an open op and `session-retired` for a retired
+  one. The runner gets refused and exits 21. Retire rechecks in its own transaction that no
+  `claim` is current for the session, unless it's the proven-ended run's. The no-run proof also
+  takes `lockSessionWorkspace`, which a spawned run holds before it claims.
+- **B3, op namespace.** `sessions` has `UNIQUE(thread, op)` (schema.sql:263), and the successor
+  takes the clear op. Both directions refuse `session-op-conflict` at accept, before any stop:
+  - `session.clear` refuses an op already in `sessions` or `session_ops` for the thread.
+  - `session.create` (broker.mjs:5859-5872) refuses an op already in `session_ops`, instead of
+    answering the successor row.
+- **B4** is in the schema step below (the `attempt` key).
+- **N2, name cap.** `name()` caps names at 200 characters (broker.mjs:323). When `<old>-N` would
+  be longer than that, session.clear refuses `session-name-too-long` at accept and writes nothing
+  (sage-2828). Test: a 199-character name clears to a 201-character `<old>-1` and is refused, and
+  a 198-character name clears.
+- **N3, batch.** Drain waits for the runner's whole receive batch. Messages already delivered
+  are a turn in progress, and the runner runs each of them before its next receive (:543-590).
+  Only undelivered messages are cancelled. This is a residual: a batch of k delivered messages
+  delays the drain by k turns.
+- **N4, D7 step 1 rechecks.** Retire rechecks owner, write grant and halt in its transaction, and
+  refuses to `failed` with the code. External effects: the old run's endRun record is the stop
+  evidence the op joins (its run and generation are on `drained`). No other effect is reconciled
+  in 271.
+- **N5, claim end.** Retire ends the old claim only if that run's `claim` is still the latest
+  `thread_claims` row for the session. If endRun already ended it, nothing is written twice. Test:
+  retire after endRun, with exactly one end row.
+- **N6, pid 1.** `--as-pid-1` is in `NAMESPACES` (seat/src/sandbox.mjs:68), and `sandboxArgv`
+  uses it (:168) for the runner command that session.mjs:302 builds. At build I'll confirm that
+  thread runs reach session.mjs:302. Teardown on exit is kernel behaviour, and the test stays.
+- **N7**: Duckworth checks the mos-1315 list line by line at the code round.
+- **N8**: row 206 is approved at 3l→3m, so if it lands first, 271 is 3m→3n (as mos-1315 item 4
+  says).
+
+Cancel tests (sage-2828, duckworth-0171 C2):
+- before retirement (no drained yet; and drained with the claim held)
+- a cancel racing the drained receive, each order
+- a cancel after `start-held`: no further report, and threadPass doesn't start the successor
+- held, held, then started on one op gives three start rows with reports 1-3 and one work
+  attempt (C1)
+- after retirement, with no further start
+- an idempotent repeat
+- a changed payload refused
+- after completion: `session-clear-complete`
+- an agent caller refused
+
+Mutant: drop the fence release in cancel's pre-retire path. The kill is the pre-retire test's next
+send/receive.
+
+Tests gained: B1 a cancel at arrival (reply chain and spec notice) for clearing and retired; B2
+a claim refused while clearing, and retire refusing a live claim; B3 both op directions; B4 a
+second `start-held` appended; N2; N5.
+
+duckworth-0171 N1: the `cancelled` key on message.send is additive. At build I'll grep the
+whole-answer `deepEqual` pins on message.send in `packages/*/tests/*`. A pin breaks only where a
+test's routed message lands on a clearing or retired session, which no test does before 271.
+Any hit goes into the mos-1315 list first.
+
+### Schema step (label by landing order, mos-1315 item 4)
+
+- `session_ops(seq, at, business, thread, session, op, stage, attempt, successor, successor_name,
+  run, generation, digest, refusal)`:
+  - append-only, with `UNIQUE(thread, op, stage, attempt)` and a CHECK on `stage`
+  - the stages are `requested`, `drained`, `uncertain`, `exited`, `retired`, `failed`,
+    `cancelled`, `start-held`, `start-started` and `start-refused`
+  - `attempt` is 0 for every stage up to `retired`, and for `failed` and `cancelled`. For the
+    three start stages it is a report counter: 1 plus the number of start rows the op already
+    has. Every `sessionClearStart` report bumps it, a held report included. This keys D7's
+    retryable start (duckworth-0170 B4, duckworth-0171 C1).
+  - The launcher's work attempt goes in its own `work_attempt` column, null outside the start
+    stages. A held start spends none (D8, t1-sessions.md:274), so held, held, started share one
+    work attempt and take reports 1, 2 and 3.
+  - The stage CHECK pins `attempt = 0` for every non-start stage, `cancelled` included.
+  - `session_ops_one_open` treats an op as open while its latest stage is `requested`, `drained`,
+    `uncertain` or `exited`. `retired`, `failed` and `cancelled` close it, so a new clear on the
+    same session can follow a cancelled or failed one. A cancelled op stays in the namespace (B3).
+  - a trigger refuses a second open op per session
+  - the envelope comes from the latest stage, and `revision` is the count of this session's rows
+- `message_cancels(seq, at, message UNIQUE, session, op)`, append-only.
+- `threadRows`'s `pending` gains `AND NOT EXISTS (cancel)`, and each row gains `state`.
+- Label: 3m if 271 lands before row 206. If 206 lands first (3l to 3m at Launchpad's 631a256ba),
+  271 rebases on it and is 3m to 3n. Sage sets the order.
+
+### Pre-existing test changes (mos-1315)
+
+These are allowed:
+- the current schema label
+- the fabricated future step in comment-migration.test.mjs, moved up one label
+- the migration file lists, which gain the new step's file
+
+No assertion is removed or loosened, and no other expected value changes. This list was measured
+on 2d0d407ba with `git grep` for `'3l'`, `3k-3l` and `'3m'` in `packages/*/tests/*`.
+
+- Current label (`'3l'` becomes the new label):
+  - account-migration.test.mjs:94, :144, :163, plus the rewrite source :156
+  - chief-of-staff.test.mjs:641, plus the rewrite source :617
+  - comment-migration.test.mjs:36
+  - gitea-migration.test.mjs:98, :124, :142, plus the rewrite source :135
+  - grant-migration.test.mjs:103, :128, :146, plus the rewrite source :139
+  - name-migration.test.mjs:68
+  - sessions.test.mjs:377, :450, plus the rewrite source :433
+  - spec-migration.test.mjs:66
+  - thread-migration.test.mjs:104, plus the rewrite source :52
+  - usage-migration.test.mjs:71
+  - A rewrite source is the `.replace("export const SCHEMA_VERSION = '3l';", ...)` string. It has
+    to match store.mjs:17's new text, or the rewrite silently does nothing.
+- Fabricated future step: comment-migration.test.mjs:78-79, :86 (`3l`→`3m` becomes one label up).
+- Migration file lists:
+  - comment-migration.test.mjs:82
+  - the `copyFileSync` lists that copy `3k-3l.sql`: account-migration.test.mjs:86,
+    gitea-migration.test.mjs:76, grant-migration.test.mjs:79
+  - plus each one's sibling copy loop, read whole at the edit
+- sessions.test.mjs:435 asserts which step names the edited source holds. If its expected list
+  needs the new step's file, that falls under the migration-file-list allowance. I'll read it
+  whole at the edit.
+- **Restore lists (mos-1318 item 1).** Both new tables have an `at` column, so both join
+  `COUNTED` in scripts/stack-release.sh:420. That is product scope for 271: a rollback with
+  `--restore` counts their rows written after the backup and refuses on them, as it does for
+  `sessions` and `thread_opens`. T46 (stack-release.test.mjs:913) then passes unchanged. Neither
+  table goes in `untimed`, and nothing is removed from any list.
+  - `session_ops`: COUNTED. Every write is a clear stage, and a restore that dropped one uncounted
+    could leave a retired session's successor without its journal.
+  - `message_cancels`: COUNTED. A dropped cancel would put a cancelled message back in `pending`.
+  - T45's `LISTED` (stack-release.test.mjs:858) is a hand copy of COUNTED, so that a table dropped
+    from the script fails T45. **Both names are added** (mos-1319). Mos counts this as the
+    stricter change mos-1285 allows, because T45's loop then checks each new table on its own.
+    The reviewer checks that LISTED matches COUNTED name for name: 15 names after 271, where
+    there were 13. On this base COUNTED is at stack-release.sh:420. Mos measured it at :418 on
+    aca16404c.
+- **Count pins (mos-1318 item 2).** schema.test.mjs:525-527 (table, trigger and view counts):
+  - tables 22 to 24: `session_ops`, `message_cancels`
+  - triggers 86 to 93:
+    - `session_ops_no_update`, `session_ops_no_delete`, `session_ops_no_replace`
+    - `session_ops_one_open` (refuses a second open op on a session)
+    - `message_cancels_no_update`, `message_cancels_no_delete`, `message_cancels_no_replace`
+  - views: 5, unchanged
+  - The comment block above the pins gains a row 271 line, as rows 124-264a have.
+  - If the build adds or drops an object, this list changes first and the reviewer checks it.
+  - If row 206 (631a256ba) lands first, 271 re-adds its own objects on top of 206's counts.
+  - Any other count pin found at build goes here before it's edited.
+- After the move, grep `packages/*/tests` for the old label. Only history fixtures and migration
+  chains may still name it. The result is recorded here (mos-1315 item 3).
+- Comment-only mentions (`3k-3l migration` in end-launch, message-filter, thread-claims, threads,
+  session-launch, thread-cli, thread-launch, chat, session-compactions) are history and stay.
+
+### Tests (T4, to write)
+
+- clear on a held session:
+  - accept fences the session
+  - the run's next receive appends `drained` and answers `[]`
+  - SIGTERM, then exit proof, `retired`, the successor row and the default move
+- a landing-turn run (no receive yet) is never drained or signalled
+- a run whose turn is running (delivered, unread) stays `clearing` with no signal, past 30 s
+- a non-default session clears with the default unchanged
+- `<old>-N` collision: a user session already named `<old>-1` gives `<old>-2`
+- repeat with the same payload gives the same envelope, id and name
+- conflict: a changed payload refuses `session-op-conflict`
+- a second op on a clearing session refuses `session-clearing`
+- stale revision, halted processing, retired session
+- undelivered messages are cancelled with receipts, not delivered to the successor
+- sends to a clearing session refuse
+- `uncertain` after 30 s with no proof, and no second signal
+- the successor start is held, started or refused, through check()/start()
+- the pid-namespace build check

docs/plans/DEFERRED.md

@@ -478,6 +478,8 @@ at every gate. Started 2026-09-12 during the control board MVP.
   not that the event's `actor_run` is the refused run. Binds row 269 (264b,
   #1774), or the next row that changes `sessions.test.mjs`: pin `actor_run`
   in `refusedOnce`. (2026-10-22)
+  Closed by row 269 (264b, #1774): `refusedOnce` selects `actor_run` and
+  expects the refused run's name (mos-1292). (2026-10-22)
 
 - **editor-provenance skips its rebuild check on esbuild 0.28.1.** (Duckworth
   N2 on the row 264a manifest round, duckworth-0146.) The webui test

packages/bus/README.md

@@ -244,7 +244,7 @@ a receipt with empty `changes`. A refusal is `action.refused` with its code alon
 | `thread.open` | `{thread,session?}`; human sessions only. Records a `thread_opens` row, which asks the launcher to start the thread's session. Returns `{opened}`. |
 | `thread.send` | `{thread,session?,body,class?}`; human sessions only. A message to the thread's instance, in that thread, with its `human.input` event as `request`. Returns `{id,request}`. |
 | `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). Writes nothing. |
-| `session.list` | `{thread}`; human and tester sessions (row 264a, #1769). The thread's sessions oldest first, `[{id,at,name,predecessor,default}]`. Writes nothing. |
+| `session.list` | `{thread}`; human and tester sessions (row 264a, #1769). The thread's sessions oldest first, `[{id,at,name,predecessor,default,compactions}]`. `compactions` is `{count,known,reason,at}` from the host's ledger (row 270, #1775): `known` true gives the exact count of completed Pi compactions; otherwise `count` is a lower bound or null and `reason` is `harness-unsupported`, `not-collected`, `store-unreadable` or `store-incomplete`. Writes nothing. |
 | `session.create` | `{thread,name,op}`; human and tester sessions. A new session of the thread, metadata only. A retry with the same `op` and name returns the first row; another name under that `op` refuses `session-op-conflict`, and a taken name `session-exists`. Returns `{id,thread,name}`. |
 
 ### Projects and threads (row 121, #1615)

packages/bus/schema.sql

@@ -478,3 +478,40 @@ CREATE TABLE gitea_grants (
   PRIMARY KEY (business, agent),
   FOREIGN KEY (business, agent) REFERENCES gitea_users(business, agent)
 ) STRICT;
+CREATE TABLE session_ops (
+  seq INTEGER PRIMARY KEY AUTOINCREMENT,
+  at TEXT NOT NULL,
+  business TEXT NOT NULL,
+  thread TEXT NOT NULL REFERENCES threads(id),
+  session TEXT NOT NULL REFERENCES sessions(id),
+  op TEXT NOT NULL,
+  stage TEXT NOT NULL CHECK (stage IN ('requested','drained','uncertain','exited','retired','failed','cancelled','start-held','start-started','start-refused')),
+  attempt INTEGER NOT NULL CHECK (CASE WHEN stage GLOB 'start-*' THEN attempt >= 1 ELSE attempt = 0 END),
+  work_attempt INTEGER CHECK (work_attempt IS NULL OR (stage GLOB 'start-*' AND work_attempt >= 1)),
+  successor TEXT NOT NULL,
+  successor_name TEXT NOT NULL,
+  run TEXT,
+  generation INTEGER CHECK (generation IS NULL OR generation >= 1),
+  digest TEXT,
+  refusal TEXT,
+  UNIQUE (thread, op, stage, attempt)
+) STRICT;
+CREATE TABLE message_cancels (
+  seq INTEGER PRIMARY KEY AUTOINCREMENT,
+  at TEXT NOT NULL,
+  message TEXT NOT NULL UNIQUE REFERENCES messages(id),
+  session TEXT NOT NULL REFERENCES sessions(id),
+  op TEXT NOT NULL
+) STRICT;
+CREATE INDEX session_ops_session ON session_ops (session, seq);
+CREATE TRIGGER session_ops_no_update BEFORE UPDATE ON session_ops BEGIN SELECT RAISE(ABORT, 'session_ops is append-only'); END;
+CREATE TRIGGER session_ops_no_delete BEFORE DELETE ON session_ops BEGIN SELECT RAISE(ABORT, 'session_ops is append-only'); END;
+CREATE TRIGGER session_ops_no_replace BEFORE INSERT ON session_ops WHEN EXISTS (SELECT 1 FROM session_ops WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'session_ops is append-only'); END;
+CREATE TRIGGER session_ops_one_open BEFORE INSERT ON session_ops
+  WHEN NEW.stage = 'requested' AND EXISTS (SELECT 1 FROM session_ops o WHERE o.session = NEW.session
+    AND o.seq = (SELECT max(seq) FROM session_ops WHERE thread = o.thread AND op = o.op)
+    AND o.stage IN ('requested','drained','uncertain','exited'))
+  BEGIN SELECT RAISE(ABORT, 'session has an open clear'); END;
+CREATE TRIGGER message_cancels_no_update BEFORE UPDATE ON message_cancels BEGIN SELECT RAISE(ABORT, 'message_cancels is append-only'); END;
+CREATE TRIGGER message_cancels_no_delete BEFORE DELETE ON message_cancels BEGIN SELECT RAISE(ABORT, 'message_cancels is append-only'); END;
+CREATE TRIGGER message_cancels_no_replace BEFORE INSERT ON message_cancels WHEN EXISTS (SELECT 1 FROM message_cancels WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'message_cancels is append-only'); END;

packages/bus/src/broker.mjs

@@ -384,6 +384,7 @@ export class Broker {
   // FED R3: () => the checked access document (R1's loadAccess), read once per user request. Null,
   // a throw or no document leaves every user with no grant (design 3.5, fail closed).
   #access;
+  #compactions;
   // r6 X3: interrupted host grants whose remote undo waits for reconcileHosts, each with its claim.
   #hostUndo = [];
   // r6 section 6 (N4): revokes a restart cut short, whose step 3 waits for reconcileHosts.
@@ -409,13 +410,14 @@ export class Broker {
   #clock = 0;
   #processing;
   // credentialStatus is the runtime's metadata-only Credentials.status(); the broker never gets use().
-  constructor({ store, businesses, secretCheck = () => {}, credentialStatus = () => [], organization = null, project = null, library = null, usage = null, limits = null, harnessOn = null, credentialVerbs = null, hostGrant = null, hostKeys = null, processing = null, testerOn = null, roleFile = null, giteaGrant = null, giteaKeys = null, access = null }) {
+  constructor({ store, businesses, secretCheck = () => {}, credentialStatus = () => [], organization = null, project = null, library = null, usage = null, limits = null, harnessOn = null, credentialVerbs = null, hostGrant = null, hostKeys = null, processing = null, testerOn = null, roleFile = null, giteaGrant = null, giteaKeys = null, access = null, compactions = null }) {
     if (!object(businesses) || !Object.keys(businesses).length) fail('invalid-business');
     // Row 194 b15 (sage-2501): the function-argument checks come before anything reads the store.
     if (harnessOn !== null && typeof harnessOn !== 'function') throw new TypeError('harnessOn must be a function');
     if (testerOn !== null && typeof testerOn !== 'function') throw new TypeError('testerOn must be a function');
     if (roleFile !== null && typeof roleFile !== 'function') throw new TypeError('roleFile must be a function');
     if (access !== null && typeof access !== 'function') throw new TypeError('access must be a function');
+    if (compactions !== null && typeof compactions !== 'function') throw new TypeError('compactions must be a function');
     this.#businesses = structuredClone(businesses);
     this.#store = store;
     // A Broker is a trusted host startup, including in-process hosts. Never repair an existing state as off.
@@ -435,6 +437,7 @@ export class Broker {
     this.#giteaGrant = giteaGrant;
     this.#giteaKeys = giteaKeys;
     this.#access = access;
+    this.#compactions = compactions;
     this.#project = project;
     this.#library = library;
     this.#usage = usage;
@@ -752,6 +755,15 @@ export class Broker {
     if (s.tester && this.#store.get('SELECT owner FROM threads WHERE id=?', thread).owner !== this.#owner(s)) fail('thread-not-owned');
     return thread;
   }
+  #compactionCount(t, session, legacy) {
+    const unknown = (reason) => ({ count: null, known: false, reason, at: null });
+    if (!this.#compactions) return unknown('not-collected');
+    try {
+      return this.#compactions({ business: t.business, thread: t.id, session, legacy });
+    } catch {
+      return unknown('store-unreadable');
+    }
+  }
   #thread(s, id) {
     const t = this.#store.get('SELECT * FROM threads WHERE id=? AND business=?', identifier(id), s.business);
     if (!t) fail('thread-not-found');
@@ -837,7 +849,7 @@ export class Broker {
     return cap;
   }
   // Row 121 (#1615): what the launcher needs to start, stop and recover thread sessions.
-  // Trusted launcher API, not a socket verb. Per thread: messages to the instance that no
+  // Trusted launcher API, not a socket verb. Per session of each thread (row 269, 264b): messages to the instance that no
   // run has been delivered yet, messages delivered to the current holder and not yet read
   // (a turn in progress), and the latest message, open and claim times. A message delivered
   // to an earlier run and never read (it crashed mid-turn) isn't the holder's turn, so it
@@ -857,30 +869,31 @@ export class Broker {
     const session = this.#runs.get(key) ?? fail('unknown-run');
     if (this.#ended.has(key)) fail('run-ended');
     // A run holds at most one session, so this also refuses another thread and a non-thread run.
-    // Row 264a: the row is the thread's default session's, so a run of another session is refused too.
-    const [t] = this.#threadRows('t.business=? AND t.id=?', record.business, identifier(record.thread));
+    // Row 269 (264b): the row is the session the run was bound to, so a run is idled against its own.
+    const [t] = this.#threadRows('t.business=? AND t.id=? AND se.id=?', record.business, identifier(record.thread), session.session);
     if (t?.holder !== record.run) fail('not-holder');
     if (t.pending || t.unread) return { released: false, pending: t.pending, unread: t.unread };
     this.#store.transaction(() => this.#claimEnd(session, this.#own(session), 'release', null));
     return { released: true };
   }
-  // Row 264a (#1769): each row is the thread's default session's. Until 264b the launcher starts
-  // default sessions only, so another session's messages wait.
+  // Row 269 (264b, #1774): one row per session of each thread, oldest thread first, then oldest
+  // session. `legacy` is a session the 3k-3l migration made (op legacy-3l), whose store and
+  // in-sandbox workspace path are the thread's old ones.
   #threadRows(where, ...params) {
     return this.#store
       .all(
-        `SELECT t.id, t.instance,
-          (SELECT count(*) FROM messages m JOIN message_threads mt ON mt.message=m.id WHERE mt.session=ds.session AND m.to_role=t.instance
+        `SELECT t.id, t.instance, se.id AS session, se.op AS session_op,
+          (SELECT count(*) FROM messages m JOIN message_threads mt ON mt.message=m.id WHERE mt.session=se.id AND m.to_role=t.instance
             AND NOT EXISTS(SELECT 1 FROM deliveries d WHERE d.message=m.id AND d.op IN ('delivered','read'))) AS pending,
-          (SELECT count(*) FROM messages m JOIN message_threads mt ON mt.message=m.id WHERE mt.session=ds.session AND m.to_role=t.instance
+          (SELECT count(*) FROM messages m JOIN message_threads mt ON mt.message=m.id WHERE mt.session=se.id AND m.to_role=t.instance
             AND c.op='claim' AND EXISTS(SELECT 1 FROM deliveries d WHERE d.message=m.id AND d.op='delivered' AND d.holder_run=c.holder_run)
             AND NOT EXISTS(SELECT 1 FROM deliveries d WHERE d.message=m.id AND d.op='read')) AS unread,
-          (SELECT max(m.at) FROM messages m JOIN message_threads mt ON mt.message=m.id WHERE mt.session=ds.session) AS last_message,
-          (SELECT max(at) FROM thread_opens WHERE session=ds.session) AS last_open,
+          (SELECT max(m.at) FROM messages m JOIN message_threads mt ON mt.message=m.id WHERE mt.session=se.id) AS last_message,
+          (SELECT max(at) FROM thread_opens WHERE session=se.id) AS last_open,
           c.at AS claim_at, c.op AS claim_op, c.holder_run AS holder_run
-        FROM threads t JOIN session_defaults ds ON ds.seq=(SELECT max(seq) FROM session_defaults WHERE thread=t.id)
-          LEFT JOIN thread_claims c ON c.seq=(SELECT max(seq) FROM thread_claims WHERE session=ds.session)
-        WHERE ${where} ORDER BY t.seq`,
+        FROM threads t JOIN sessions se ON se.thread=t.id
+          LEFT JOIN thread_claims c ON c.seq=(SELECT max(seq) FROM thread_claims WHERE session=se.id)
+        WHERE ${where} ORDER BY t.seq, se.seq`,
         ...params,
       )
       .map((r) => ({
@@ -892,6 +905,8 @@ export class Broker {
         lastOpen: r.last_open,
         lastClaim: r.claim_at,
         holder: r.claim_op === 'claim' ? r.holder_run : null,
+        session: r.session,
+        legacy: r.session_op === 'legacy-3l',
       }));
   }
   // Trusted launcher API, deliberately NOT a socket verb. S6 supplies immutable launch records.
@@ -5826,14 +5841,17 @@ export class Broker {
       }
       // Row 264a (#1769, T1 item 7): a thread's sessions, oldest first, and which one is the default.
       // Each user reaches only their own threads' sessions (REQ-USER-2); there are no shares (Q7).
+      // Row 270 (264c, D6): each row's compaction count, from the host's ledger reader (the bus
+      // host's seat compactions.mjs; this file is mounted in the sandbox and can't import it).
+      // Without a reader the count is unknown, never 0.
       case 'session.list': {
         if (!s.tester) this.#human(s);
         keys(a, ['thread'], ['thread']);
         const t = this.#thread(s, a.thread);
         const fallback = this.#defaultSession(t.id);
         return this.#store
-          .all('SELECT id,at,name,predecessor FROM sessions WHERE thread=? ORDER BY seq', t.id)
-          .map((r) => ({ ...r, default: r.id === fallback }));
+          .all('SELECT id,at,name,predecessor,op FROM sessions WHERE thread=? ORDER BY seq', t.id)
+          .map(({ op, ...r }) => ({ ...r, default: r.id === fallback, compactions: this.#compactionCount(t, r.id, op === 'legacy-3l') }));
       }
       // D8: a new session is metadata only. Its run starts when the launcher starts it (264b).
       // The op key makes a retry find the first row: the same op and name answer it, another name

packages/bus/src/migrations/3l-3m.sql

@@ -0,0 +1,40 @@
+-- Row 271 (264d, #1776, T1 item 7 D7/D8): clear context. session_ops is the clear journal, one row
+-- per stage, and message_cancels holds a receipt for each message a clear cancelled. Only new
+-- tables, an index and their triggers.
+CREATE TABLE session_ops (
+  seq INTEGER PRIMARY KEY AUTOINCREMENT,
+  at TEXT NOT NULL,
+  business TEXT NOT NULL,
+  thread TEXT NOT NULL REFERENCES threads(id),
+  session TEXT NOT NULL REFERENCES sessions(id),
+  op TEXT NOT NULL,
+  stage TEXT NOT NULL CHECK (stage IN ('requested','drained','uncertain','exited','retired','failed','cancelled','start-held','start-started','start-refused')),
+  attempt INTEGER NOT NULL CHECK (CASE WHEN stage GLOB 'start-*' THEN attempt >= 1 ELSE attempt = 0 END),
+  work_attempt INTEGER CHECK (work_attempt IS NULL OR (stage GLOB 'start-*' AND work_attempt >= 1)),
+  successor TEXT NOT NULL,
+  successor_name TEXT NOT NULL,
+  run TEXT,
+  generation INTEGER CHECK (generation IS NULL OR generation >= 1),
+  digest TEXT,
+  refusal TEXT,
+  UNIQUE (thread, op, stage, attempt)
+) STRICT;
+CREATE TABLE message_cancels (
+  seq INTEGER PRIMARY KEY AUTOINCREMENT,
+  at TEXT NOT NULL,
+  message TEXT NOT NULL UNIQUE REFERENCES messages(id),
+  session TEXT NOT NULL REFERENCES sessions(id),
+  op TEXT NOT NULL
+) STRICT;
+CREATE INDEX session_ops_session ON session_ops (session, seq);
+CREATE TRIGGER session_ops_no_update BEFORE UPDATE ON session_ops BEGIN SELECT RAISE(ABORT, 'session_ops is append-only'); END;
+CREATE TRIGGER session_ops_no_delete BEFORE DELETE ON session_ops BEGIN SELECT RAISE(ABORT, 'session_ops is append-only'); END;
+CREATE TRIGGER session_ops_no_replace BEFORE INSERT ON session_ops WHEN EXISTS (SELECT 1 FROM session_ops WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'session_ops is append-only'); END;
+CREATE TRIGGER session_ops_one_open BEFORE INSERT ON session_ops
+  WHEN NEW.stage = 'requested' AND EXISTS (SELECT 1 FROM session_ops o WHERE o.session = NEW.session
+    AND o.seq = (SELECT max(seq) FROM session_ops WHERE thread = o.thread AND op = o.op)
+    AND o.stage IN ('requested','drained','uncertain','exited'))
+  BEGIN SELECT RAISE(ABORT, 'session has an open clear'); END;
+CREATE TRIGGER message_cancels_no_update BEFORE UPDATE ON message_cancels BEGIN SELECT RAISE(ABORT, 'message_cancels is append-only'); END;
+CREATE TRIGGER message_cancels_no_delete BEFORE DELETE ON message_cancels BEGIN SELECT RAISE(ABORT, 'message_cancels is append-only'); END;
+CREATE TRIGGER message_cancels_no_replace BEFORE INSERT ON message_cancels WHEN EXISTS (SELECT 1 FROM message_cancels WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'message_cancels is append-only'); END;

packages/bus/src/runtime.mjs

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

packages/bus/src/store.mjs

@@ -13,8 +13,8 @@ import { join, isAbsolute } from 'node:path';
 import { createHash } from 'node:crypto';
 
 const ddl = readFileSync(new URL('../schema.sql', import.meta.url), 'utf8');
-export const SCHEMA_SHA256 = 'cd68512212ce8bf8f5631138b2e63a340fa7b940088430649bb7784af7df162a';
-export const SCHEMA_VERSION = '3l';
+export const SCHEMA_SHA256 = '129a76db0fe9973cdea5a87f274941b4111c86730fba8a0cbe16e61a534525d0';
+export const SCHEMA_VERSION = '3m';
 // Ordered, named offline steps. New schema changes append a step and keep prior DDL pinned.
 const legacyDDL = readFileSync(new URL('./migrations/3b.sql', import.meta.url), 'utf8');
 const LEGACY_SHA256 = '179ffe356d4ff19a49b5ebad39b6c6bfd7771deb8e1b7c55b5e746f039d69e65';
@@ -72,6 +72,11 @@ const migrations = [{
   from: '3k',
   to: '3l',
   sql: readFileSync(new URL('./migrations/3k-3l.sql', import.meta.url), 'utf8'),
+}, {
+  // Row 271 (#1776): clear context. The clear journal and the cancelled-message receipts. Only new tables.
+  from: '3l',
+  to: '3m',
+  sql: readFileSync(new URL('./migrations/3l-3m.sql', import.meta.url), 'utf8'),
 }];
 const timedTables = [
   'events',
@@ -89,6 +94,8 @@ const timedTables = [
   'thread_opens',
   'instance_names',
   'usage_records',
+  'session_ops',
+  'message_cancels',
 ];
 const hex = (value) => createHash('sha256').update(value).digest('hex');
 if (hex(ddl) !== SCHEMA_SHA256) throw Error('schema-source-mismatch');

packages/bus/tests/end-launch.test.mjs

@@ -191,7 +191,7 @@ test('broker process: launch ops authorize role.launch, record refusals and end
   assert.deepEqual((await ask(child, { op: 'credentialStatus', business: 'demo', role: 'pm' })).result, []);
   // The launcher's thread view, through the same child.
   assert.deepEqual((await ask(child, { op: 'threadsState', business: 'demo' })).result, [
-    { thread: 'th-1', instance: 'coder', pending: 0, unread: 0, lastMessage: null, lastOpen: '2026-10-15T00:00:01.000Z', lastClaim: null, holder: null },
+    { thread: 'th-1', instance: 'coder', pending: 0, unread: 0, lastMessage: null, lastOpen: '2026-10-15T00:00:01.000Z', lastClaim: null, holder: null, session: 'th-1', legacy: true },
   ]);
   // And its idle release (Dewey's F3, 28591): the op reaches the broker, which refuses a run
   // that doesn't hold the thread.

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

@@ -0,0 +1,98 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { randomUUID } from 'node:crypto';
+import { mkdtempSync, readFileSync, rmSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { fileURLToPath } from 'node:url';
+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 shipped = (name) =>
+  JSON.parse(readFileSync(fileURLToPath(new URL(`../../../roles/${name}.json`, import.meta.url)), 'utf8'));
+const businesses = {
+  demo: {
+    id: 'demo',
+    human: 'jason',
+    arbiters: { technical: 'pm', delivery: 'chief' },
+    roles: {
+      chief: { definition: 'chief-of-staff', authority: { withinRole: ['message.send'], crossRole: [] } },
+      assistant: { definition: 'assistant', authority: { withinRole: ['message.send'], crossRole: [] } },
+      pm: { definition: 'pm', authority: { withinRole: ['message.send'], crossRole: [] } },
+    },
+    definitions: { 'chief-of-staff': {}, assistant: { messaging: shipped('assistant').messaging }, 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, project, 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, its id the thread's) keeps the
+// thread's store. The thread is made by hand with the session the migration gives it, as
+// message-filter.test.mjs does; sessions is append-only.
+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 }) });
+  const L = randomUUID();
+  const at = '2026-10-21T00:00:00.000Z';
+  f.store.run('INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)', L, at, 'demo', f.project, 'assistant', 'jason', 'legacy');
+  f.store.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);
+  f.store.run("INSERT INTO session_defaults(at,thread,session,by) SELECT at,id,id,'migration-3l' FROM threads WHERE id=?", L);
+  f.call('session.list', { thread: L });
+  assert.deepEqual(asked, [{ business: 'demo', thread: L, session: L, legacy: true }]);
+});
+
+test('without a reader the count is unknown, never 0; a reader that throws is store-unreadable', (t) => {
+  const f = setup(t);
+  assert.deepEqual(f.call('session.list', { thread: f.A }).map((r) => r.compactions), [{ count: null, known: false, reason: 'not-collected', at: null }]);
+  const g = setup(t, {
+    compactions: () => {
+      throw new Error('EACCES');
+    },
+  });
+  assert.deepEqual(g.call('session.list', { thread: g.A }).map((r) => r.compactions), [{ count: null, known: false, reason: 'store-unreadable', at: null }]);
+});
+
+test('compactions must be a function', (t) => {
+  const root = mkdtempSync(join(tmpdir(), 'bus-compactions-'));
+  const store = new Store(root);
+  t.after(() => {
+    store.close();
+    rmSync(root, { recursive: true, force: true });
+  });
+  assert.throws(() => new Broker({ store, businesses, compactions: {} }), { name: 'TypeError', message: 'compactions must be a function' });
+});

packages/bus/tests/sessions.test.mjs

@@ -199,21 +199,21 @@ test('T2 sage-2257: a released or revoked run\'s retired generation refuses its
   // Each refusal leaves its one action.refused event (request()'s catch, #refused); nothing else is
   // written: no approval, delivery, read or claim state (mos-1277).
   const lastSeq = () => store.get('SELECT max(seq) AS s FROM events').s;
-  const refusedOnce = (fn, label) => {
+  const refusedOnce = (fn, label, run) => {
     const from = lastSeq();
     assert.throws(fn, { code: 'not-holder' }, label);
-    const added = store.all('SELECT kind, body FROM events WHERE seq > ? ORDER BY seq', from);
+    const added = store.all('SELECT kind, actor_run, body FROM events WHERE seq > ? ORDER BY seq', from);
     assert.deepEqual(
-      added.map((e) => [e.kind, JSON.parse(e.body)]),
-      [['action.refused', { code: 'not-holder' }]],
+      added.map((e) => [e.kind, e.actor_run, JSON.parse(e.body)]),
+      [['action.refused', run, { code: 'not-holder' }]],
       label,
     );
   };
   const refused = (name, { d, taken }) => {
     const { events, ...before } = counts();
-    refusedOnce(() => call(name, 'decision.withdraw', { id: d.id }), 'approval');
-    refusedOnce(() => call(name, 'message.receive'), 'delivery');
-    refusedOnce(() => call(name, 'message.read', { id: taken.id }), 'read');
+    refusedOnce(() => call(name, 'decision.withdraw', { id: d.id }), 'approval', `${name}-run`);
+    refusedOnce(() => call(name, 'message.receive'), 'delivery', `${name}-run`);
+    refusedOnce(() => call(name, 'message.read', { id: taken.id }), 'read', `${name}-run`);
     const { events: after, ...rest } = counts();
     assert.deepEqual(rest, before);
     assert.equal(after, events + 3);

packages/cli/src/launcher.mjs

@@ -75,6 +75,15 @@
 // threadStore()), so the next run for the thread resumes the harness
 // session (REQ-THREAD-2).
 //
+// Row 269 (264b, #1774): threadsState has one row per session of each
+// thread, and everything above is per session. Each session has its own
+// store (session.mjs sessionStore()) and its own workspace under
+// session-workspaces/ (sessionWorkspace()), which a lock beside it keeps to
+// one writer from before the clones until the run's proven exit. A legacy
+// session (op legacy-3l) keeps the thread's store, and its new workspace
+// shows inside the sandbox at the old instance path, the cwd its history
+// recorded. The broker's half of the fence is its per-session claim.
+//
 // Landing runs (row 156, #1651). Every launch of a non-thread instance
 // reads the broker's landHolders (launch op): the approved, unspent merge
 // decisions each non-thread instance raised. The instance's entries go into
@@ -121,12 +130,14 @@
 //   processing-halted.
 //   - delete: instance-running against the registry, then lifecycleDelete
 //     (the writer's commit), then the rename of workspaces/<b>/I/ to
-//     workspaces/<b>/.retired/I-<transaction>/. A failed rename closes it
-//     workspace-rename-failed with the transaction; a later delete retries it.
+//     workspaces/<b>/.retired/I-<transaction>/, and of
+//     session-workspaces/<b>/I/ to session-workspaces/<b>/.retired/I-<transaction>/
+//     (row 269). A failed rename closes it workspace-rename-failed with the
+//     transaction; a later delete retries it.
 
 import { createServer } from "node:net";
 import { chmodSync, existsSync, lstatSync, mkdirSync, readFileSync, renameSync, rmSync } from "node:fs";
-import { dirname, join } from "node:path";
+import { dirname, join, relative } from "node:path";
 import { configDir, loadBusiness, resolveInstance, systemVars } from "../../business/src/index.mjs";
 import { limitHit } from "../../business/src/launch.mjs";
 import { BundleError, buildBundle, hasFallback, readSkills, sessionModel } from "../../harness/src/bundle.mjs";
@@ -142,13 +153,16 @@ import {
   runnerOf,
   SESSION_CREDENTIALS,
   sessionEnv,
-  threadStore,
   sessionExit,
+  sessionStore,
+  sessionWorkspace,
+  lockSessionWorkspace,
   spawnSession,
   workspaceDir,
   writeSessionFile,
 } from "../../seat/src/session.mjs";
 import { BWRAP, sandboxProblem } from "../../seat/src/sandbox.mjs";
+import { collectCompactions } from "../../seat/src/compactions.mjs";
 import { CheckoutError, ensureCheckouts } from "../../seat/src/checkout.mjs";
 import { createGitSocket } from "./git-socket.mjs";
 import { cmdlineOf, startTimeOf } from "../../seat/src/proc.mjs";
@@ -237,6 +251,9 @@ export function createLauncher({
     throw new CliError(`${e.message}; check which sessions still run, then move the file aside`, 3);
   }
   const registry = new Registry(dataRoot, { readOnly: !launches });
+  // Row 269: earlier-host runs still live after recover(). check() sees each while it lives.
+  const survivors = [];
+  const liveRun = (s) => startTimeOf(s.pid) === s.startTime || startTimeOf(s.runnerPid) === s.runnerStartTime;
   const children = new Map();
   let chain = Promise.resolve();
   let closed = false;
@@ -267,6 +284,22 @@ export function createLauncher({
     }
   };
 
+  // Row 270 (264c, D6): count a thread session's completed compactions into its
+  // store's ledger, at run start (what earlier runs and a resume left) and at
+  // run end. The caller holds the session's workspace lock. A failure is
+  // logged and recorded, and never stops or fails the run: the count then
+  // reads unknown with its reason.
+  const compactions = (store, harness, { instance, thread, session, run }) => {
+    let c;
+    try {
+      c = collectCompactions(store, harness);
+    } catch (e) {
+      log(`run ${run}: compactions: ${e.message}`);
+      c = { count: null, known: false, reason: "store-unreadable" };
+    }
+    record({ event: "compactions", instance, thread, session, run, count: c.count, known: c.known, reason: c.reason });
+  };
+
   function serialize(fn) {
     const run = chain.then(fn);
     chain = run.catch(() => {});
@@ -278,7 +311,7 @@ export function createLauncher({
   // nothing else launches for one. Row 194 (#1692, V2): `human` is the
   // human's launch request, which may name any declared instance. The listing
   // limits the chief, not the human. launch.by still takes the `pm` branch.
-  async function check(instance, { pm = false, thread = null, human = false } = {}) {
+  async function check(instance, { pm = false, thread = null, session = null, human = false } = {}) {
     const admissionGeneration = admission().generation;
     try { b=loadBusiness(host.business,{dir:configDir(env),rolesDir:rolesDir(env)}); } catch(e){throw new Refusal("organization-refused",e.message);}
     if (!b.launch) throw new Refusal("launch-not-configured", `business ${b.id} has no launch block`);
@@ -286,7 +319,12 @@ export function createLauncher({
     const threaded = b.roles[instance]?.definition === THREAD_DEFINITION;
     if (threaded && !thread) throw new Refusal("thread-required", `${instance} runs only for a thread`);
     if (!threaded && thread) throw new Refusal("thread-not-allowed", `${instance} isn't a thread instance`);
-    if (registry.running(b.id, instance, thread)) throw new Refusal("instance-running");
+    if (threaded && !session) throw new Refusal("session-required", `${instance} runs for one session of a thread`);
+    if (registry.running(b.id, instance, thread, session)) throw new Refusal("instance-running");
+    // Row 269 (Q2): a run an earlier host started with no session works in the instance's old
+    // workspace and the thread's store, so no session of the instance starts while one lives.
+    if (threaded && [...registry.sessions, ...leftover, ...survivors.filter(liveRun)].some((s) => s.business === b.id && s.instance === instance && !s.session))
+      throw new Refusal("legacy-run-live", `a run of ${instance} with no session is still live`);
     let resolved;
     try {
       resolved = resolveInstance({ system: vars, business: b, instance });
@@ -378,29 +416,59 @@ export function createLauncher({
   }
 
   // Bundle, spawn, bind, register, hand over the capability.
-  async function start({ instance, launchedBy, resolved, model, account = null, thread = null, human = false, admissionGeneration }) {
+  async function start({ instance, launchedBy, resolved, model, account = null, thread = null, session = null, legacy = false, legacyThread = legacy, human = false, admissionGeneration }) {
     admission(admissionGeneration);
     // Every launch path ends here, so a --no-launch host spawns nothing whichever path asked.
     if (!launches) throw new Refusal("launches-disabled", "this host was started with --no-launch");
-    // A thread's store is checked first, so a bad id or an unsafe directory
-    // launches nothing.
+    // A thread session's store is checked first, so a bad id or an unsafe
+    // directory launches nothing.
     let store = null;
     if (thread) {
       try {
-        store = threadStore(dataRoot, b.id, thread);
+        store = sessionStore(dataRoot, b.id, thread, session, { legacy, legacyThread });
       } catch (e) {
-        throw new Refusal("thread-store-refused", e.message);
+        throw new Refusal(e.code === "legacy-store-unmapped" ? e.code : "thread-store-refused", e.message);
       }
     }
     const credentials = await host.op({ op: "credentialStatus", business: b.id, role: instance });
     admission(admissionGeneration);
     // The clones come before the run, so one that can't be made leaves no
     // run behind. The host's git treats each clone as hostile, and its
-    // checks hold only while no session writes the workspace: a workspace
-    // is one instance's, check() refused instance-running, and launches are
-    // serialized, so none does.
-    const workspace = workspaceDir(dataRoot, b.id, instance);
-    mkdirSync(workspace, { recursive: true, mode: 0o700 });
+    // checks hold only while no session writes the workspace. A non-thread
+    // workspace is one instance's, check() refused instance-running, and
+    // launches are serialized. A thread session's workspace is its own (row
+    // 269), and its lock refuses a second writer, whichever host asks.
+    let workspace;
+    let workspaceDest = null;
+    let lock = null;
+    if (thread) {
+      try {
+        const live = registry.sessions.filter((s) => s.business === b.id && s.workspace).map((s) => s.workspace);
+        workspace = sessionWorkspace(dataRoot, b.id, instance, session, { live });
+        lock = lockSessionWorkspace(workspace, { hostPid: process.pid, hostStartTime: startTimeOf(process.pid), business: b.id, instance, thread, session });
+      } catch (e) {
+        if (["session-workspace-unsafe", "session-workspace-overlap", "session-workspace-busy"].includes(e.code)) throw new Refusal(e.code, e.message);
+        throw e;
+      }
+      if (legacy) workspaceDest = workspaceDir(dataRoot, b.id, instance);
+    } else {
+      workspace = workspaceDir(dataRoot, b.id, instance);
+      mkdirSync(workspace, { recursive: true, mode: 0o700 });
+    }
+    try {
+      return await startLocked({ instance, launchedBy, resolved, model, account, thread, session, human, admissionGeneration, store, credentials, workspace, workspaceDest, lock });
+    } catch (e) {
+      // Every failure path in startLocked has proven the run's exit, or never spawned one.
+      lock?.release();
+      throw e;
+    }
+  }
+
+  // The rest of start(): clones, bundle, spawn, bind, register. The lock, when there is one, is
+  // released at the run's close.
+  async function startLocked({ instance, launchedBy, resolved, model, account, thread, session, human, admissionGeneration, store, credentials, workspace, workspaceDest, lock }) {
+    // What the session sees: a legacy session's workspace at the old instance path.
+    const inSandbox = workspaceDest ?? workspace;
     let checkouts;
     try {
       checkouts = await withProcessing(dataRoot, () => {
@@ -416,11 +484,12 @@ export function createLauncher({
       else if (c.outcome === "fetched") record({ event: "checkout-fetched", instance, project: c.project, reason: c.reason });
     }
     let harness, route, routeReason;
-    ({resolved,model,account,harness,route,routeReason}=await check(instance,{pm:instance===b.launch.by,thread,human}));
+    ({resolved,model,account,harness,route,routeReason}=await check(instance,{pm:instance===b.launch.by,thread,session,human}));
     admission(admissionGeneration);
     const { run, runDir } = newRun(dataRoot, b.id);
     const sessionDir = store ? store.session : join(runDir, "session");
     if (!store) mkdirSync(sessionDir, { mode: 0o700 });
+    if (store) compactions(store, model.harness, { instance, thread, session, run });
     const toolSocket = join(runDir, "tools.sock");
     const turnMarker = join(runDir, "turn.done");
     const version = harnessVersion !== undefined ? harnessVersion : call(model.harness, "version", () => harness.version({ env }));
@@ -440,7 +509,7 @@ export function createLauncher({
       arbiter: Object.values(b.arbiters).includes(instance),
       tracker,
       credentials,
-      session: { run, launchedBy, workspace, toolSocket, turnMarker, checkouts },
+      session: { run, launchedBy, workspace: inSandbox, toolSocket, turnMarker, checkouts: inSandbox === workspace ? checkouts : checkouts.map((c) => ({ ...c, path: join(inSandbox, relative(workspace, c.path)) })) },
       harness,
       route,
       routeReason,
@@ -471,7 +540,7 @@ export function createLauncher({
       brokerSocket: host.path,
       launchSocket: launchSocketPath(dataRoot),
       gitSocket: git,
-      workspace,
+      workspace: inSandbox,
       sessionDir,
       adapter,
       bundle: files,
@@ -484,6 +553,7 @@ export function createLauncher({
 
     const paths = {
       workspace,
+      workspaceDest,
       adapter,
       harnessMounts,
       brokerSocket: host.path,
@@ -519,11 +589,12 @@ export function createLauncher({
       updateRuntime(dataRoot, managed.id, { runnerPid: runner.pid, runnerStartTime: runner.startTime });
       admission(admissionGeneration);
       if (!outerStart) throw new Error("the session process exited at once");
-      const launch = await host.bindLaunch({ business: b.id, role: instance, run, harness: model.harness, provider: model.provider, model: model.model, account, transcript, pid: child.pid, startTime: outerStart, ...(thread ? { thread } : {}) });
+      lock?.update({ run, runPid: child.pid, runStartTime: outerStart });
+      const launch = await host.bindLaunch({ business: b.id, role: instance, run, harness: model.harness, provider: model.provider, model: model.model, account, transcript, pid: child.pid, startTime: outerStart, ...(thread ? { thread, session } : {}) });
       bound = true;
       admission(admissionGeneration);
       entry = {
-        business: b.id, instance, ...(thread ? { thread } : {}), run, launchedBy, harness: model.harness, model: model.model, provider: model.provider, modelKey: model.modelKey,
+        business: b.id, instance, ...(thread ? { thread, session, workspace } : {}), run, launchedBy, harness: model.harness, model: model.model, provider: model.provider, modelKey: model.modelKey,
         pid: child.pid, startTime: outerStart, runnerPid: runner.pid, runnerStartTime: runner.startTime, runDir, startedAt: new Date().toISOString(),
       };
       registry.add(entry);
@@ -552,13 +623,16 @@ export function createLauncher({
         const ended = await endRun(run, reason, signal ? null : code);
         registry.remove(b.id, run);
         // Row 125 (#1619, REQ-LIMIT-4): a run a limit check stopped names the limit, from the claim row.
-        record({ event: "end", instance, ...(thread ? { thread } : {}), run, reason, exitCode: code, signal, ...(haltGeneration ? { haltGeneration } : {}), ...(ended?.limit ? { limit: ended.limit } : {}) });
+        record({ event: "end", instance, ...(thread ? { thread, session } : {}), run, reason, exitCode: code, signal, ...(haltGeneration ? { haltGeneration } : {}), ...(ended?.limit ? { limit: ended.limit } : {}) });
         log(`run ${run} (${instance}) ended: ${reason}`);
       } catch (e) {
         log(`run ${run}: ${e.message}`);
       } finally {
         try { unregisterRuntime(dataRoot, managed.id); }
         catch { log(`run ${run}: processing receipt cleanup refused`); }
+        if (store) compactions(store, model.harness, { instance, thread, session, run });
+        try { lock?.release(); }
+        catch (e) { log(`run ${run}: workspace lock release: ${e.message}`); }
         children.delete(run);
         stopping.delete(run);
         spawnedAt.delete(run);
@@ -566,7 +640,7 @@ export function createLauncher({
         refusedRelease.delete(run);
       }
     });
-    record({ event: "launch", instance, ...(thread ? { thread } : {}), run, launchedBy, harness: model.harness, route, routeReason, model: model.model, provider: model.provider, modelKey: model.modelKey, account, pid: child.pid, runnerPid: entry.runnerPid, manifest });
+    record({ event: "launch", instance, ...(thread ? { thread, session } : {}), run, launchedBy, harness: model.harness, route, routeReason, model: model.model, provider: model.provider, modelKey: model.modelKey, account, pid: child.pid, runnerPid: entry.runnerPid, manifest });
     return { run, instance, harness: model.harness, model: model.model };
   }
 
@@ -600,9 +674,12 @@ export function createLauncher({
     }
     const now = Date.now();
     const ms = (at) => (at ? Date.parse(at) : 0);
+    // Row 269: a thread with a legacy session keeps its old store for that session only.
+    const legacyThreads = new Set(threads.filter((t) => t.legacy).map((t) => t.thread));
     for (const t of threads) {
       if (closed) return;
-      const s = registry.running(b.id, t.instance, t.thread);
+      const key = `${t.thread}/${t.session}`;
+      const s = registry.running(b.id, t.instance, t.thread, t.session);
       if (s) {
         const last = Math.max(ms(s.startedAt), ms(t.lastMessage), ms(t.lastOpen));
         if (!stopping.has(s.run) && now - last >= b.launch.threadIdleSeconds * 1000) {
@@ -616,7 +693,7 @@ export function createLauncher({
           // A turn in progress, or a message since the snapshot: the run keeps going.
           if (!idle.released) continue;
           stopping.add(s.run);
-          record({ event: "thread-idle", instance: t.instance, thread: t.thread, run: s.run });
+          record({ event: "thread-idle", instance: t.instance, thread: t.thread, session: t.session, run: s.run });
           try {
             if (startTimeOf(s.runnerPid) === s.runnerStartTime) process.kill(s.runnerPid, "SIGTERM");
           } catch {}
@@ -629,20 +706,21 @@ export function createLauncher({
       // the work (it died before its claim, say) waits for the next one
       // rather than start again every tick. A refused start isn't a start.
       const signature = `${t.lastMessage}|${t.lastOpen}`;
-      if (startedFor.get(t.thread) === signature) continue;
+      if (startedFor.get(key) === signature) continue;
       await serialize(async () => {
-        if (closed || registry.running(b.id, t.instance, t.thread)) return;
+        if (closed || registry.running(b.id, t.instance, t.thread, t.session)) return;
         try {
-          await start({ instance: t.instance, launchedBy: b.human, thread: t.thread, ...(await check(t.instance, { thread: t.thread })) });
-          startedFor.set(t.thread, signature);
-          refusedThreads.delete(t.thread);
+          const where = { thread: t.thread, session: t.session };
+          await start({ instance: t.instance, launchedBy: b.human, ...where, legacy: t.legacy, legacyThread: legacyThreads.has(t.thread), ...(await check(t.instance, where)) });
+          startedFor.set(key, signature);
+          refusedThreads.delete(key);
         } catch (e) {
           const code = e instanceof Refusal ? e.code : "launch-failed";
           // Logged once per code, not once per tick.
-          if (refusedThreads.get(t.thread) === code) return;
-          refusedThreads.set(t.thread, code);
-          log(`thread ${t.thread} (${t.instance}): start refused: ${code}`);
-          record({ event: "refused", by: b.human, instance: t.instance, thread: t.thread, code, why: e.message, ...(e.limit?{limit:e.limit}:{}) });
+          if (refusedThreads.get(key) === code) return;
+          refusedThreads.set(key, code);
+          log(`thread ${t.thread} session ${t.session} (${t.instance}): start refused: ${code}`);
+          record({ event: "refused", by: b.human, instance: t.instance, thread: t.thread, session: t.session, code, why: e.message, ...(e.limit?{limit:e.limit}:{}) });
         }
       });
     }
@@ -849,20 +927,23 @@ export function createLauncher({
     const transaction = out?.transaction ?? null;
     if (transaction === null) return { refusal: "workspace-rename-failed" };
     const bot = out.bot ? { bot: out.bot } : {};
-    const from = workspaceDir(dataRoot, b.id, r.instance);
-    if (!existsSync(from)) return { transaction, ...bot };
-    const to = join(dirname(from), ".retired", `${r.instance}-${transaction}`);
-    try {
-      mkdirSync(dirname(to), { recursive: true, mode: 0o700 });
-      chmodSync(dirname(to), 0o700);
-      if (existsSync(to)) throw new Error(`${to} exists`);
-      renameSync(from, to);
-      chmodSync(to, 0o700);
-    } catch (e) {
-      log(`request ${r.request} (${r.instance}): workspace rename failed: ${e.message}; left at ${from}`);
-      return { transaction, ...bot, refusal: "workspace-rename-failed", why: `${e.message}; left at ${from}` };
+    // The old instance workspace, then (row 269) the instance's session workspaces, each moved
+    // whole. A retry moves whichever is still in place.
+    for (const from of [workspaceDir(dataRoot, b.id, r.instance), join(dataRoot, "session-workspaces", b.id, r.instance)]) {
+      if (!existsSync(from)) continue;
+      const to = join(dirname(from), ".retired", `${r.instance}-${transaction}`);
+      try {
+        mkdirSync(dirname(to), { recursive: true, mode: 0o700 });
+        chmodSync(dirname(to), 0o700);
+        if (existsSync(to)) throw new Error(`${to} exists`);
+        renameSync(from, to);
+        chmodSync(to, 0o700);
+      } catch (e) {
+        log(`request ${r.request} (${r.instance}): workspace rename failed: ${e.message}; left at ${from}`);
+        return { transaction, ...bot, refusal: "workspace-rename-failed", why: `${e.message}; left at ${from}` };
+      }
+      record({ event: "workspace-retired", instance: r.instance, request: r.request, transaction, path: to });
     }
-    record({ event: "workspace-retired", instance: r.instance, request: r.request, transaction, path: to });
     return { transaction, ...bot };
   }
 
@@ -958,10 +1039,15 @@ export function createLauncher({
           process.kill(s.pid, "SIGKILL");
         } catch {}
         const end = Date.now() + 5000;
-        while (startTimeOf(s.pid) === s.startTime && Date.now() < end) await new Promise((r) => setTimeout(r, 20));
+        while (liveRun(s) && Date.now() < end) await new Promise((r) => setTimeout(r, 20));
+      }
+      // Row 269 (Q2): the session or its runner outlived the kill, or wasn't a session to kill.
+      if (liveRun(s)) {
+        survivors.push(s);
+        log(`run ${s.run} (${s.instance}) from an earlier host still runs (pid ${s.pid}, runner ${s.runnerPid}): no session of ${s.instance} starts until it exits`);
       }
       try {
-        await host.bindLaunch({ business: b.id, role: s.instance, run: s.run, harness: s.harness, pid: s.pid, startTime: s.startTime, ...(s.thread ? { thread: s.thread } : {}) });
+        await host.bindLaunch({ business: b.id, role: s.instance, run: s.run, harness: s.harness, pid: s.pid, startTime: s.startTime, ...(s.thread ? { thread: s.thread } : {}), ...(s.session ? { session: s.session } : {}) });
       } catch (e) {
         // A host that died after ending the run but before updating sessions.json.
         if (e.code === "run-ended") log(`run ${s.run} (${s.instance}) from an earlier host was already ended`);
@@ -969,7 +1055,7 @@ export function createLauncher({
         continue;
       }
       await endRun(s.run, "host-lost", null);
-      record({ event: "end", instance: s.instance, ...(s.thread ? { thread: s.thread } : {}), run: s.run, reason: "host-lost", exitCode: null, signal: null });
+      record({ event: "end", instance: s.instance, ...(s.thread ? { thread: s.thread } : {}), ...(s.session ? { session: s.session } : {}), run: s.run, reason: "host-lost", exitCode: null, signal: null });
       log(`run ${s.run} (${s.instance}) from an earlier host ended: host-lost`);
     }
     leftover = [];

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

@@ -0,0 +1,18 @@
+// A stand-in harness turn for the session launch tests (row 269, #1774).
+// thread-adapter.mjs's turn, plus the run's cwd and the names in it, so a
+// test reads from the reply which workspace the session saw and at what
+// in-sandbox path. It keeps a history in $MOSAIC_SESSION_DIR, the session's
+// store.
+
+import { appendFileSync, existsSync, readdirSync, 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 cwd = process.cwd();
+const entries = readdirSync(cwd).sort();
+appendFileSync(historyFile, `${body}\n`);
+process.stdout.write(`THREAD ${JSON.stringify({ body, before, cwd, entries })}\n`);
+writeFileSync(process.env.MOSAIC_TURN_MARKER, "done\n");

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

@@ -0,0 +1,414 @@
+// Session runs (row 269, 264b, #1774): every session of a thread runs on its
+// own, with its own store under threads/<b>/<t>/sessions/<s>/ and its own
+// workspace under session-workspaces/<b>/<instance>/<s>/, and a lock beside
+// the workspace keeps it to one writer. A legacy session (op legacy-3l) keeps
+// the thread's old store, and its workspace shows inside the sandbox at the
+// old instance path. Real runners in the bwrap sandbox; a stub adapter keeps
+// a history in the store and answers with what it found, its cwd and what
+// the cwd holds.
+//
+// The host is thread-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 { spawn, spawnSync } from "node:child_process";
+import { chmodSync, existsSync, mkdirSync, readdirSync, readFileSync, symlinkSync, 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 { transition } from "../../processing-control/src/state.mjs";
+import { gitEnv } from "../../seat/src/checkout.mjs";
+import { startTimeOf } from "../../seat/src/proc.mjs";
+import { sandboxProblem } from "../../seat/src/sandbox.mjs";
+import { launchLogFile, readSessions, Registry, SESSION_CREDENTIALS } 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";
+// A thread made by hand gets the default session the 3k-3l migration gives a legacy thread, with
+// the thread's id (as in thread-launch.test.mjs).
+const DEFAULT_SESSION = [
+  "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=?",
+  "INSERT INTO session_defaults(at,thread,session,by) SELECT at,id,id,'migration-3l' FROM threads WHERE id=?",
+];
+
+const ADAPTER = fileURLToPath(new URL("./fixtures/session-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 the waits a mutant could
+// stall return what they last saw and the assertions after them decide.
+async function settle(fn, ms = 20000) {
+  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);
+}
+
+// Each directory made 0700, as the store and workspace checks want.
+function privateDir(...parts) {
+  for (let i = 1; i <= parts.length; i++) {
+    const d = join(...parts.slice(0, i));
+    if (!existsSync(d)) mkdirSync(d, { mode: 0o700 });
+    if (i > 1) chmodSync(d, 0o700);
+  }
+  return join(...parts);
+}
+
+// The assistant is the thread instance; sonnet is its model's limit.
+function prepare(t, { sonnet = 3, idle = 1 } = {}) {
+  const root = tmp(t, "mosaic-session-");
+  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 });
+    doc.launch.threadIdleSeconds = idle;
+    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. `ops` replaces launch ops, given the runtime.
+async function start(t, { f, adapter }, { tickMs = 100, ops: replace } = {}) {
+  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),
+    ...replace?.(runtime),
+  };
+  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: tickMs,
+    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 { host, 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 sessionWs = (f, session) => join(f.dataRoot, "session-workspaces", "acme", "assistant", session);
+// The refusals once there are n, or once anything launched.
+const outcome = async (f, n = 1) => {
+  await settle(() => events(f, "refused").length >= n || events(f, "launch").length);
+  return events(f, "refused").map((e) => [e.session, e.code]);
+};
+const historyOf = (p) => (existsSync(p) ? readFileSync(p, "utf8") : null);
+
+// Send to one session of a thread; reply() waits for the assistant's RESULT reply in that session.
+function send(call, thread, session, body) {
+  const sent = call("thread.send", { thread, ...(session ? { session } : {}), body });
+  const reply = async () => {
+    const r = await until(() => call("thread.messages", { thread, ...(session ? { session } : {}) }).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]);
+  };
+  return { sent, reply };
+}
+
+// A thread made with thread.create, its default session and a second named one.
+function twoSessions(call) {
+  const project = call("project.create", { name: "stack work", repo: "stack" }).id;
+  const A = call("thread.create", { project, name: "alpha" }).id;
+  const [{ id: D }] = call("session.list", { thread: A });
+  const S = call("session.create", { thread: A, name: "second", op: "op-second" }).id;
+  return { project, A, D, S };
+}
+
+test("REQ-THREAD-2 REQ-LAUNCH-1 two sessions of one thread, and another thread's default, run at once, each with its own store and workspace; a restart of one resumes only that one", { skip, timeout: 90000 }, async (t) => {
+  const prepared = prepare(t, { idle: 5 });
+  const { f } = prepared;
+  const host = await start(t, prepared);
+  const { project, A, D, S } = twoSessions(host.call);
+  const B = host.call("thread.create", { project, name: "beta" }).id;
+  const [{ id: BD }] = host.call("session.list", { thread: B });
+
+  const toD = send(host.call, A, D, "first for D");
+  const toS = send(host.call, A, S, "first for S");
+  const toB = send(host.call, B, null, "first for B");
+  // All three run together (M1, M2: a shared workspace would refuse the second start busy).
+  await settle(() => readSessions(f.dataRoot).length === 3 || events(f, "refused").length);
+  assert.deepEqual(events(f, "refused"), []);
+  assert.deepEqual(readSessions(f.dataRoot).map((s) => [s.thread, s.session]).sort(), [[A, D], [A, S], [B, BD]].sort());
+  const [sawD, sawS, sawB] = await Promise.all([toD.reply(), toS.reply(), toB.reply()]);
+  // Each session's history is its own (M5), and each runs in its own workspace at its own path (M8).
+  assert.deepEqual([sawD.body, sawD.before, sawD.cwd], ["first for D", [], sessionWs(f, D)]);
+  assert.deepEqual([sawS.body, sawS.before, sawS.cwd], ["first for S", [], sessionWs(f, S)]);
+  assert.deepEqual([sawB.body, sawB.before, sawB.cwd], ["first for B", [], sessionWs(f, BD)]);
+  const store = (thread, session) => join(f.dataRoot, "threads", "acme", thread, "sessions", session, "session", "history");
+  assert.deepEqual([historyOf(store(A, D)), historyOf(store(A, S)), historyOf(store(B, BD))], ["first for D\n", "first for S\n", "first for B\n"]);
+  for (const thread of [A, B]) assert.equal(existsSync(join(f.dataRoot, "threads", "acme", thread, "session")), false);
+  assert.equal(existsSync(join(f.dataRoot, "workspaces", "acme", "assistant")), false, "no thread session uses the old instance workspace");
+  for (const e of events(f, "launch")) assert.ok([D, S, BD].includes(e.session), JSON.stringify(e));
+
+  // All three stop when idle, and each lock goes with its run.
+  await until(() => events(f, "end").length === 3, 30000);
+  assert.deepEqual(readSessions(f.dataRoot), []);
+  for (const s of [D, S, BD]) assert.equal(existsSync(`${sessionWs(f, s)}.lock`), false, s);
+
+  // The named session's next message starts only that session (M9), which finds only its history.
+  const again = await send(host.call, A, S, "second for S").reply();
+  assert.deepEqual([again.before, again.cwd], [["first for S"], sessionWs(f, S)]);
+  assert.deepEqual(events(f, "launch").slice(3).map((e) => [e.thread, e.session]), [[A, S]]);
+  assert.equal(historyOf(store(A, D)), "first for D\n");
+});
+
+// D9 (M7): a thread from before 3l. Its legacy session resumes from the thread's old store, and
+// its new workspace shows at the old instance path, which the transcripts recorded as the cwd.
+// The old host workspace isn't mounted.
+test("REQ-THREAD-2 REQ-LAUNCH-1 a legacy session resumes the thread's old store at the old in-sandbox cwd, in a workspace of its own", { skip, timeout: 60000 }, async (t) => {
+  const prepared = prepare(t);
+  const { f } = prepared;
+  const store = new Store(f.dataRoot);
+  const at = () => new Date().toISOString();
+  store.run("INSERT INTO projects(id,at,business,name,repo,created_by) VALUES(?,?,?,?,?,?)", "p-hand", at(), "acme", "hand made", "stack", "jason");
+  store.run("INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)", "th-legacy", at(), "acme", "p-hand", "assistant", "jason", "old");
+  for (const sql of DEFAULT_SESSION) store.run(sql, "th-legacy");
+  store.close();
+  const old = privateDir(f.dataRoot, "threads", "acme", "th-legacy");
+  writeFileSync(join(privateDir(old, "session"), "history"), "old sentinel\n");
+  privateDir(old, "claude-projects");
+  writeFileSync(join(privateDir(f.dataRoot, "workspaces", "acme", "assistant"), "old-file"), "the 264a workspace\n");
+
+  const host = await start(t, prepared);
+  const saw = await send(host.call, "th-legacy", null, "after 3l").reply();
+  assert.deepEqual(saw.before, ["old sentinel"]);
+  assert.equal(saw.cwd, join(f.dataRoot, "workspaces", "acme", "assistant"));
+  assert.equal(saw.entries.includes("old-file"), false, "the old host workspace isn't mounted");
+  assert.equal(readFileSync(join(old, "session", "history"), "utf8"), "old sentinel\nafter 3l\n");
+  assert.equal(existsSync(join(old, "sessions")), false);
+  assert.ok(existsSync(sessionWs(f, "th-legacy")));
+  assert.deepEqual(readdirSync(join(f.dataRoot, "workspaces", "acme", "assistant")), ["old-file"]);
+  assert.deepEqual(events(f, "launch").map((e) => [e.thread, e.session]), [["th-legacy", "th-legacy"]]);
+});
+
+// T2, M3: another host's live lock on the session's workspace refuses the start before any
+// checkout. Once that host is gone the lock is stale, and the next start takes it over.
+test("REQ-THREAD-2 REQ-LAUNCH-1 a live lock on a session workspace refuses session-workspace-busy before any checkout; a stale one is taken over", { skip, timeout: 60000 }, async (t) => {
+  const prepared = prepare(t);
+  const { f } = prepared;
+  const host = await start(t, prepared);
+  const { A, S } = twoSessions(host.call);
+  const other = spawn("sleep", ["60"], { stdio: "ignore" });
+  const gone = new Promise((r) => other.once("exit", r));
+  t.after(() => other.kill("SIGKILL"));
+  const otherStart = await until(() => startTimeOf(other.pid));
+  privateDir(f.dataRoot, "session-workspaces", "acme", "assistant");
+  const lock = `${sessionWs(f, S)}.lock`;
+  const held = `${JSON.stringify({ hostPid: other.pid, hostStartTime: otherStart, business: "acme", instance: "assistant", thread: A, session: S })}\n`;
+  writeFileSync(lock, held, { mode: 0o600 });
+
+  const toS = send(host.call, A, S, "for S");
+  assert.deepEqual(await outcome(f), [[S, "session-workspace-busy"]]);
+  assert.equal(events(f, "refused")[0].thread, A);
+  assert.deepEqual(events(f, "launch"), []);
+  assert.deepEqual(readSessions(f.dataRoot), []);
+  assert.deepEqual(readdirSync(sessionWs(f, S)), [], "no checkout");
+  assert.equal(readFileSync(lock, "utf8"), held);
+
+  other.kill("SIGKILL");
+  await gone;
+  const saw = await toS.reply();
+  assert.deepEqual([saw.before, saw.cwd], [[], sessionWs(f, S)]);
+  await until(() => events(f, "end").length === 1);
+  assert.equal(existsSync(lock), false);
+});
+
+// Q2 (BUILD.md:177-179, t1-sessions.md:109): an earlier host's sessionless run of the instance is
+// still live, so no session of it starts until that run has exited. A sleep child stands in for
+// its runner; its outer pid is gone, so recover() has nothing to kill and keeps it as a survivor.
+test("REQ-THREAD-2 REQ-LAUNCH-1 a session start refuses legacy-run-live while an earlier host's sessionless run lives, and starts once it has exited", { skip, timeout: 60000 }, async (t) => {
+  const prepared = prepare(t);
+  const { f } = prepared;
+  const runner = spawn("sleep", ["60"], { stdio: "ignore" });
+  const gone = new Promise((r) => runner.once("exit", r));
+  t.after(() => runner.kill("SIGKILL"));
+  const runnerStartTime = await until(() => startTimeOf(runner.pid));
+  new Registry(f.dataRoot).add({ business: "acme", instance: "assistant", thread: "th-old", run: "run-left", harness: "pi", pid: 2 ** 22 + 7, startTime: "gone", runnerPid: runner.pid, runnerStartTime });
+  const host = await start(t, prepared);
+  assert.ok(host.logs.includes(`run run-left (assistant) from an earlier host still runs (pid ${2 ** 22 + 7}, runner ${runner.pid}): no session of assistant starts until it exits`), host.logs.join("\n"));
+  const { A, S } = twoSessions(host.call);
+
+  const toS = send(host.call, A, S, "for S");
+  assert.deepEqual(await outcome(f), [[S, "legacy-run-live"]]);
+  assert.equal(events(f, "refused")[0].thread, A);
+  assert.deepEqual(events(f, "launch"), [], "nothing spawned");
+  assert.deepEqual(readSessions(f.dataRoot), []);
+  assert.equal(existsSync(sessionWs(f, S)), false, "refused before the workspace");
+
+  runner.kill("SIGKILL");
+  await gone;
+  assert.equal(await settle(() => events(f, "launch").length), 1, "it starts once the run has exited");
+  const saw = await toS.reply();
+  assert.deepEqual([saw.before, saw.cwd], [[], sessionWs(f, S)]);
+  await until(() => events(f, "end").length === 1);
+});
+
+// T2, C3: a symlinked session root refuses session-workspace-unsafe, at the segment check.
+test("REQ-THREAD-2 REQ-LAUNCH-1 a symlinked session workspace root refuses session-workspace-unsafe and launches nothing", { skip, timeout: 60000 }, async (t) => {
+  const prepared = prepare(t);
+  const { f } = prepared;
+  const elsewhere = privateDir(prepared.root, "elsewhere");
+  symlinkSync(elsewhere, join(f.dataRoot, "session-workspaces"));
+  const host = await start(t, prepared);
+  const { A, D } = twoSessions(host.call);
+  send(host.call, A, D, "for D");
+  assert.deepEqual(await outcome(f), [[D, "session-workspace-unsafe"]]);
+  assert.deepEqual(events(f, "launch"), []);
+  assert.deepEqual(readdirSync(elsewhere), []);
+});
+
+// T2, M4: every segment of the session path is a real directory, but workspaces/<b> is a symlink
+// to it, so only the realpath compare finds that the session workspace holds the legacy root.
+test("REQ-THREAD-2 REQ-LAUNCH-1 a session workspace that holds the legacy workspace root refuses session-workspace-overlap and launches nothing", { skip, timeout: 60000 }, async (t) => {
+  const prepared = prepare(t);
+  const { f } = prepared;
+  const host = await start(t, prepared);
+  const { A, D } = twoSessions(host.call);
+  const ws = privateDir(f.dataRoot, "session-workspaces", "acme", "assistant", D);
+  symlinkSync(ws, join(privateDir(f.dataRoot, "workspaces"), "acme"));
+  send(host.call, A, D, "for D");
+  assert.deepEqual(await outcome(f), [[D, "session-workspace-overlap"]]);
+  assert.deepEqual(events(f, "launch"), []);
+  assert.deepEqual(readdirSync(ws), []);
+});
+
+// T2: a symlink among the store's segments refuses thread-store-refused before the workspace.
+test("REQ-THREAD-2 REQ-LAUNCH-1 a symlinked session store refuses thread-store-refused and makes no workspace", { skip, timeout: 60000 }, async (t) => {
+  const prepared = prepare(t);
+  const { f } = prepared;
+  const host = await start(t, prepared);
+  const { A, D } = twoSessions(host.call);
+  const elsewhere = privateDir(prepared.root, "elsewhere");
+  symlinkSync(elsewhere, join(privateDir(f.dataRoot, "threads", "acme", A), "sessions"));
+  send(host.call, A, D, "for D");
+  assert.deepEqual(await outcome(f), [[D, "thread-store-refused"]]);
+  assert.match(events(f, "refused")[0].why, /is unsafe$/);
+  assert.deepEqual(events(f, "launch"), []);
+  assert.deepEqual(readdirSync(elsewhere), []);
+  assert.equal(existsSync(join(f.dataRoot, "session-workspaces")), false);
+});
+
+// T2, M6 and the halt fence (C2). The credentialStatus op runs in start() after check() returned.
+// Its first call halts and resumes, a new generation with no halt, so only start()'s own
+// admission(admissionGeneration) refuses, before the session workspace is made. On the retry,
+// the recheck's mayLaunch op (inside check(), after the lock is taken) halts once it has answered,
+// and the recheck's admission() refuses processing-halted. Neither attempt spawns, and the second releases its lock.
+test("REQ-THREAD-2 REQ-LAUNCH-1 a named session's start refuses processing-generation-changed, then processing-halted, after check() and before any spawn", { skip, timeout: 60000 }, async (t) => {
+  const prepared = prepare(t);
+  const { f } = prepared;
+  let credentials = 0;
+  let halts = 0;
+  const halt = () => transition(f.dataRoot, { business: "acme", human: f.doc.human, halted: true });
+  const host = await start(t, prepared, {
+    ops: (runtime) => ({
+      credentialStatus: (m) => {
+        if (m.role === "assistant" && ++credentials === 1) {
+          transition(f.dataRoot, { business: "acme", human: f.doc.human, halted: false, generation: halt().generation });
+          halts++;
+        }
+        return runtime.credentialStatus(m.business, m.role);
+      },
+      // The halt lands after the op answers: halted inside it, the bus's own fence
+      // (runtime.mjs processingWork) throws and the launcher refuses limit-check-failed.
+      mayLaunch: async (m) => {
+        const may = await runtime.mayLaunch(m);
+        if (m.instance === "assistant" && credentials === 2 && halts === 1) {
+          halt();
+          halts++;
+        }
+        return may;
+      },
+    }),
+  });
+  const { A, S } = twoSessions(host.call);
+  send(host.call, A, S, "for S");
+  const refused = await outcome(f, 2);
+  assert.deepEqual(refused, [[S, "processing-generation-changed"], [S, "processing-halted"]]);
+  assert.deepEqual([credentials, halts], [2, 2], "check() passed both times");
+  assert.equal(events(f, "launch").length, 0);
+  assert.deepEqual(readSessions(f.dataRoot), []);
+  assert.equal(existsSync(join(f.dataRoot, "launches", "acme")), false, "no run directory");
+  assert.ok(existsSync(sessionWs(f, S)), "the retry made the workspace");
+  assert.equal(existsSync(`${sessionWs(f, S)}.lock`), false, "and released its lock");
+});

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

@@ -179,10 +179,11 @@ test("REQ-THREAD-2 REQ-LAUNCH-1 REQ-MSG-1 a thread message starts the thread's s
   assert.deepEqual(first.saw, { body: "first for A", before: [], transcripts: [] });
   const [run1] = launchLog(f).filter((e) => e.event === "launch" && e.thread === A).map((e) => e.run);
   assert.ok(run1);
-  const store = join(f.dataRoot, "threads", "acme", A);
+  const store = join(f.dataRoot, "threads", "acme", A, "sessions", host.runtime.threadsState("acme").find((x) => x.thread === A).session);
   assert.equal(readFileSync(join(store, "session", "history"), "utf8"), "first for A\n");
   assert.deepEqual(readdirSync(join(store, "claude-projects")), ["turn-1.jsonl"]);
   assert.equal(existsSync(join(f.dataRoot, "launches", "acme", run1, "session")), false, "a thread run keeps no per-run session directory");
+  assert.equal(existsSync(join(f.dataRoot, "threads", "acme", A, "session")), false, "a thread.create session keeps no thread-level store");
   await idleStop(f, host, A, run1);
 
   // A second run of the same thread finds the first run's history.

packages/seat/src/compactions.mjs

@@ -0,0 +1,239 @@
+// Row 270 (264c, #1775, T1 item 7 D6): the count of completed compactions over
+// one thread session's lifetime, across its runs and native branches.
+//
+// The host collects it from the session's native store and appends what it
+// found to a ledger, <store dir>/compactions.jsonl (0600, append-only). The
+// store dir holds the mounted session/ and claude-projects/ but is not mounted
+// itself (sandbox.mjs binds()), so the agent can't reach the ledger. The
+// launcher is its only writer, and collects while it holds the session's
+// workspace lock (session.mjs lockSessionWorkspace); each collection is one
+// O_APPEND write. The count is rebuilt from the ledger, never from a page.
+//
+// Pi (pinned 0.85.1): a completed compaction is one `type: "compaction"` entry,
+// written by SessionManager.appendCompaction after the summary exists. A start
+// or an aborted compaction writes none. A `compactionSummary` message, a
+// message inside a compaction's retainedTail and a branch_summary are not
+// compactions. Pi's createBranchedSession and forkFrom copy entries into a new
+// file of the same directory with their id and timestamp unchanged, so a
+// compaction's key is (entry id, entry timestamp) over the whole store, and the
+// stream it was first seen in is recorded beside it.
+//
+// Claude Code has no pinned compaction format yet (Vikunja 466): a session
+// any Claude run collected for is unknown with harness-unsupported, never 0.
+//
+// Ledger lines, each one JSON object with v 1:
+//   { kind: "compaction", entry, completedAt, stream, at }  once per key
+//   { kind: "cursor", stream, bytes, tail, at }  read through `bytes`; `tail`
+//     is the sha256 of the up to 4096 bytes that end there
+//   { kind: "collected", at, known, reason, harness, version }
+// A cursor lets the next collection read only what was appended. A stream
+// that shrank, changed under its cursor or went away is store-incomplete from
+// then on: entries may be lost, so the count stays a lower bound for good.
+
+import { createHash } from "node:crypto";
+import { closeSync, constants as fsc, fstatSync, lstatSync, openSync, readdirSync, readFileSync, readSync, writeSync } from "node:fs";
+import { join } from "node:path";
+import { ID_PATTERN } from "../../business/src/vocabulary.mjs";
+
+export const PI_VERSION = "0.85.1";
+export const LEDGER = "compactions.jsonl";
+const TAIL = 4096;
+const CHUNK = 1 << 20;
+const STREAM = /^[A-Za-z0-9._-]{1,200}\.jsonl$/;
+
+const digest = (buf) => createHash("sha256").update(buf).digest("hex");
+const keyOf = (entry, completedAt) => JSON.stringify([entry, completedAt]);
+const segment = (x) => typeof x === "string" && ID_PATTERN.test(x) && !x.includes("..");
+
+// The store directory session.mjs sessionStore() makes, as a path only: the
+// reader never creates it. Null for an id that isn't one.
+export function storeDir(dataRoot, business, thread, session, { legacy = false } = {}) {
+  if (![business, thread, session].every(segment)) return null;
+  const dir = join(dataRoot, "threads", business, thread);
+  return legacy ? dir : join(dir, "sessions", session);
+}
+
+function readRange(fd, from, to) {
+  const out = Buffer.alloc(to - from);
+  let off = 0;
+  while (off < out.length) {
+    const n = readSync(fd, out, off, Math.min(CHUNK, out.length - off), from + off);
+    if (n === 0) break;
+    off += n;
+  }
+  return out.subarray(0, off);
+}
+
+// One collection over a Pi session directory. `cursors` are the ledger's,
+// by stream. Returns the compactions found past each cursor, the new
+// cursors, and the problems: store-unreadable (a file that can't be read or
+// isn't a Pi session) and store-incomplete (a stream that shrank, changed or
+// went away, or a line that isn't JSON). A missing directory is empty.
+export function scanPi(dir, cursors = {}) {
+  const found = [], next = {}, problems = new Set();
+  let names;
+  try {
+    names = readdirSync(dir);
+  } catch (e) {
+    if (e.code !== "ENOENT") problems.add("store-unreadable");
+    names = [];
+  }
+  for (const stream of Object.keys(cursors)) if (!names.includes(stream)) problems.add("store-incomplete");
+  for (const name of names.filter((n) => n.endsWith(".jsonl")).sort()) {
+    const path = join(dir, name);
+    let fd;
+    try {
+      if (!STREAM.test(name) || !lstatSync(path).isFile()) throw new Error("not a regular stream file");
+      fd = openSync(path, fsc.O_RDONLY | fsc.O_NOFOLLOW);
+    } catch {
+      problems.add("store-unreadable");
+      continue;
+    }
+    try {
+      const size = fstatSync(fd).size;
+      const c = cursors[name];
+      let from = 0;
+      if (c) {
+        const tail = c.bytes > size ? null : readRange(fd, Math.max(0, c.bytes - TAIL), c.bytes);
+        if (tail && tail.length === Math.min(TAIL, c.bytes) && digest(tail) === c.tail) from = c.bytes;
+        else problems.add("store-incomplete");
+      }
+      const buf = readRange(fd, from, size);
+      // Complete lines only: a partial trailing line waits for its newline.
+      const end = buf.lastIndexOf(0x0a) + 1;
+      const lines = buf.subarray(0, end).toString("utf8").split("\n");
+      lines.pop();
+      let header = from > 0;
+      for (const line of lines) {
+        if (!line.trim()) continue;
+        let e;
+        try {
+          e = JSON.parse(line);
+        } catch {
+          e = null;
+        }
+        if (!e || typeof e !== "object" || Array.isArray(e)) {
+          problems.add("store-incomplete");
+          continue;
+        }
+        if (!header) {
+          header = true;
+          if (e.type !== "session") problems.add("store-unreadable");
+          continue;
+        }
+        if (e.type === "compaction" && typeof e.id === "string" && e.id) {
+          found.push({ entry: e.id, completedAt: typeof e.timestamp === "string" ? e.timestamp : null, stream: name });
+        }
+      }
+      const bytes = from + end;
+      next[name] = { bytes, tail: digest(readRange(fd, Math.max(0, bytes - TAIL), bytes)) };
+    } catch {
+      problems.add("store-unreadable");
+    } finally {
+      closeSync(fd);
+    }
+  }
+  return { found, cursors: next, problems: [...problems].sort() };
+}
+
+// The ledger as recorded. Throws on a ledger that can't be read. `bad` is set
+// by any line that isn't a v 1 record, a torn last line included.
+export function readLedger(dir) {
+  const events = new Map(), cursors = {};
+  let last = null, incomplete = false, unsupported = false, bad = false, torn = false, text;
+  const file = join(dir, LEDGER);
+  try {
+    if (!lstatSync(file).isFile()) throw Object.assign(new Error(`${file} isn't a regular file`), { code: "ledger-unsafe" });
+    text = readFileSync(file, { encoding: "utf8", flag: fsc.O_RDONLY | fsc.O_NOFOLLOW });
+  } catch (e) {
+    if (e.code === "ENOENT") return null;
+    throw e;
+  }
+  const lines = text.split("\n");
+  if (lines.pop() !== "") bad = torn = true;
+  for (const line of lines) {
+    let r;
+    try {
+      r = JSON.parse(line);
+    } catch {
+      r = null;
+    }
+    if (!r || typeof r !== "object" || r.v !== 1) {
+      bad = true;
+      continue;
+    }
+    if (r.kind === "compaction") {
+      const key = keyOf(r.entry, r.completedAt);
+      if (!events.has(key)) events.set(key, r);
+    } else if (r.kind === "cursor") cursors[r.stream] = { bytes: r.bytes, tail: r.tail };
+    else if (r.kind === "collected") {
+      last = r;
+      if (r.reason === "store-incomplete") incomplete = true;
+      if (r.harness !== "pi") unsupported = true;
+    } else bad = true;
+  }
+  return { events, cursors, last, incomplete, unsupported, bad, torn };
+}
+
+// What session.list shows: { count, known, reason, at }. Known only when the
+// ledger is whole, every collection ran on Pi and none lost entries. Any other
+// answer is unknown with a reason and a lower bound, or null for none: never 0.
+export function compactionCount(dir) {
+  const unknown = (reason, count = null, at = null) => ({ count: count || null, known: false, reason, at });
+  let ledger;
+  try {
+    ledger = readLedger(dir);
+  } catch {
+    return unknown("store-unreadable");
+  }
+  if (!ledger) return unknown("not-collected");
+  const count = ledger.events.size;
+  const at = ledger.last?.at ?? null;
+  if (ledger.bad) return unknown("store-incomplete", count, at);
+  if (!ledger.last) return unknown("not-collected", count);
+  if (ledger.unsupported) return unknown("harness-unsupported", count, at);
+  if (ledger.incomplete) return unknown("store-incomplete", count, at);
+  if (!ledger.last.known) return unknown(ledger.last.reason, count, at);
+  return { count, known: true, reason: null, at };
+}
+
+// One collection for a session's store ({ dir, session } from sessionStore),
+// at run start (resume) and run end, under the session's workspace lock.
+// Appends the new compactions, the moved cursors and one collected line in a
+// single O_APPEND write, then answers the count. A key already in the ledger
+// is never written again, and nothing rewrites the file.
+export function collectCompactions(store, harness, { now = () => new Date().toISOString() } = {}) {
+  const ledger = readLedger(store.dir) ?? { events: new Map(), cursors: {}, torn: false };
+  const at = now();
+  const out = [];
+  if (harness === "pi") {
+    const scan = scanPi(store.session, ledger.cursors);
+    for (const f of scan.found) {
+      const key = keyOf(f.entry, f.completedAt);
+      if (ledger.events.has(key)) continue;
+      ledger.events.set(key, f);
+      out.push({ v: 1, kind: "compaction", entry: f.entry, completedAt: f.completedAt, stream: f.stream, at });
+    }
+    for (const [stream, c] of Object.entries(scan.cursors)) {
+      const old = ledger.cursors[stream];
+      if (!old || old.bytes !== c.bytes || old.tail !== c.tail) out.push({ v: 1, kind: "cursor", stream, bytes: c.bytes, tail: c.tail, at });
+    }
+    // store-incomplete outranks store-unreadable: lost entries don't come back.
+    const reason = scan.problems.includes("store-incomplete") ? "store-incomplete" : scan.problems[0] ?? null;
+    out.push({ v: 1, kind: "collected", at, known: reason === null, reason, harness, version: PI_VERSION });
+  } else {
+    out.push({ v: 1, kind: "collected", at, known: false, reason: "harness-unsupported", harness, version: null });
+  }
+  const fd = openSync(join(store.dir, LEDGER), fsc.O_WRONLY | fsc.O_APPEND | fsc.O_CREAT | fsc.O_NOFOLLOW, 0o600);
+  try {
+    if (!fstatSync(fd).isFile()) throw Object.assign(new Error(`${join(store.dir, LEDGER)} isn't a regular file`), { code: "ledger-unsafe" });
+    // After a torn line, start on a fresh one: the torn line stays bad and
+    // keeps the count unknown, and the new lines stay whole.
+    const buf = Buffer.from((ledger.torn ? "\n" : "") + out.map((r) => JSON.stringify(r)).join("\n") + "\n");
+    let off = 0;
+    while (off < buf.length) off += writeSync(fd, buf, off, buf.length - off);
+  } finally {
+    closeSync(fd);
+  }
+  return compactionCount(store.dir);
+}

packages/seat/src/sandbox.mjs

@@ -97,9 +97,12 @@ function systemBinds(node = process.execPath) {
 // the harness module's mounts (row 65: its mounts(), [{ src, dest, why }],
 // read-only), the broker socket, the launch socket (null for a role that may
 // not launch), the git socket (null for a role that neither pushes nor
-// reviews), the thread's session store (session.mjs threadStore(), or null)
-// and node. `readOnly` adds files, for tests only.
-export function binds({ runDir, workspace, adapter, harnessMounts = [], brokerSocket, launchSocket = null, gitSocket = null, threadStore = null, node = process.execPath, readOnly = [] }) {
+// reviews), the thread's session store (session.mjs threadStore() or
+// sessionStore(), or null) and node. workspaceDest (row 269) is where the
+// workspace shows inside the sandbox when that isn't its host path: a legacy
+// thread session's, so its harness finds the cwd its history recorded.
+// `readOnly` adds files, for tests only.
+export function binds({ runDir, workspace, workspaceDest = null, adapter, harnessMounts = [], brokerSocket, launchSocket = null, gitSocket = null, threadStore = null, node = process.execPath, readOnly = [] }) {
   const out = systemBinds(node);
   for (const f of RUNNER_FILES) out.push({ kind: "ro", src: f, why: "a repository file the runner loads" });
   out.push({ kind: "ro", src: adapter, why: "the harness adapter the runner starts each turn" });
@@ -109,7 +112,7 @@ export function binds({ runDir, workspace, adapter, harnessMounts = [], brokerSo
   if (gitSocket) out.push({ kind: "ro", src: gitSocket, why: "the host's git socket, for push_working and fetch_review, and land_work" });
   for (const f of readOnly) out.push({ kind: "ro", src: f, why: "a test fixture" });
   out.push(
-    { kind: "rw", src: workspace, why: "the instance's workspace, where the harness works" },
+    { kind: "rw", src: workspace, ...(workspaceDest && workspaceDest !== workspace ? { dest: workspaceDest } : {}), why: "the instance's workspace, where the harness works" },
     { kind: "rw", src: runDir, why: "the run directory: turns/, session/, tools.sock, turn.done, home/, tmp/" },
     { kind: "ro", src: join(runDir, "bundle"), why: "the run's bundle stays as the host wrote it" },
     { kind: "ro", src: join(runDir, "session.json"), why: "the session file stays as the host wrote it" },

packages/seat/src/session.mjs

@@ -20,17 +20,22 @@
 //                                     every launch, refusal and end
 //   workspaces/<business>/<instance>/ the instance's workspace (0700), kept
 //                                     across launches
-//   threads/<business>/<thread>/      a thread's session store (0700), kept
-//                                     across the thread's runs (row 121):
+//   threads/<business>/<thread>/      a legacy thread session's store (0700),
+//                                     kept across its runs (row 121):
 //                                     session/ and claude-projects/
+//   threads/<business>/<thread>/sessions/<session>/
+//                                     any other session's store (row 269)
+//   session-workspaces/<business>/<instance>/<session>/
+//                                     a thread session's workspace (0700),
+//                                     with <session>.lock beside it
 //   bus-host/sessions.json            the running sessions (0600), written
 //                                     only by the host; `mosaic stop` reads it
 
 import { spawn } from "node:child_process";
 import { randomBytes } from "node:crypto";
-import { appendFileSync, closeSync, existsSync, lstatSync, mkdirSync, openSync, readFileSync, renameSync, writeFileSync } from "node:fs";
+import { appendFileSync, closeSync, existsSync, linkSync, lstatSync, mkdirSync, openSync, readdirSync, readFileSync, realpathSync, renameSync, unlinkSync, writeFileSync } from "node:fs";
 import { constants } from "node:os";
-import { dirname, join } from "node:path";
+import { dirname, join, sep } from "node:path";
 import { childrenOf, cmdlineOf, startTimeOf } from "./proc.mjs";
 import { binds, prepareRun, REPO, sandboxArgv } from "./sandbox.mjs";
 import { SeatError } from "./seat.mjs";
@@ -38,7 +43,10 @@ import { ID_PATTERN } from "../../business/src/vocabulary.mjs";
 
 export { REPO };
 export const RUNNER = join(REPO, "packages", "harness", "src", "runner.mjs");
-export const SESSIONS_VERSION = 1;
+// Row 269 (264b): 2 adds `session` and `workspace` to a thread run's entry.
+// readSessions still reads a version 1 file an earlier host left.
+export const SESSIONS_VERSION = 2;
+const SESSIONS_READ = [1, 2];
 
 export const sessionsFile = (dataRoot) => join(dataRoot, "bus-host", "sessions.json");
 export const launchSocketPath = (dataRoot) => join(dataRoot, "bus-host", "launch.sock");
@@ -47,6 +55,7 @@ export const gitSocketPath = (dataRoot) => join(dataRoot, "bus-host", "git.sock"
 export const runsDir = (dataRoot, business) => join(dataRoot, "launches", business);
 export const launchLogFile = (dataRoot, business) => join(dataRoot, "launches", `${business}.jsonl`);
 export const workspaceDir = (dataRoot, business, instance) => join(dataRoot, "workspaces", business, instance);
+export const sessionWorkspaceDir = (dataRoot, business, instance, session) => join(dataRoot, "session-workspaces", business, instance, session);
 
 // Row 121 (#1615): a thread's session store, which outlives each of the
 // thread's runs so the next run resumes the harness session (REQ-THREAD-2).
@@ -56,24 +65,174 @@ export const workspaceDir = (dataRoot, business, instance) => join(dataRoot, "wo
 // is the run's sessionDir; claude-projects/ is bound at the run's
 // home/.claude/projects (sandbox.mjs binds()). Throws SeatError.
 const segment = (x) => typeof x === "string" && ID_PATTERN.test(x) && !x.includes("..");
-export function threadStore(dataRoot, business, thread) {
-  for (const [what, x] of [["business", business], ["thread", thread]]) {
-    if (!segment(x)) throw new SeatError(`thread store: ${what} ${JSON.stringify(x)} isn't an id`, 2);
+const seatError = (message, code) => Object.assign(new SeatError(message, 2), { code });
+function segments(what, pairs, code) {
+  for (const [name, x] of pairs) {
+    if (!segment(x)) throw seatError(`${what}: ${name} ${JSON.stringify(x)} isn't an id`, code);
   }
-  const dir = join(dataRoot, "threads", business, thread);
-  const out = { dir, session: join(dir, "session"), projects: join(dir, "claude-projects") };
-  for (const d of [join(dataRoot, "threads"), join(dataRoot, "threads", business), dir, out.session, out.projects]) {
+}
+// Makes each directory in order, then checks it is ours, a real directory and private.
+function privateDirs(what, dirs, code) {
+  for (const d of dirs) {
     try {
       mkdirSync(d, { mode: 0o700 });
     } catch (e) {
       if (e.code !== "EEXIST") throw e;
     }
     const st = lstatSync(d);
-    if (st.isSymbolicLink() || !st.isDirectory() || st.uid !== process.getuid() || st.mode & 0o077) throw new SeatError(`thread store: ${d} is unsafe`, 2);
+    if (st.isSymbolicLink() || !st.isDirectory() || st.uid !== process.getuid() || st.mode & 0o077) throw seatError(`${what}: ${d} is unsafe`, code);
   }
+}
+export function threadStore(dataRoot, business, thread) {
+  segments("thread store", [["business", business], ["thread", thread]], "thread-store-refused");
+  const dir = join(dataRoot, "threads", business, thread);
+  const out = { dir, session: join(dir, "session"), projects: join(dir, "claude-projects") };
+  privateDirs("thread store", [join(dataRoot, "threads"), join(dataRoot, "threads", business), dir, out.session, out.projects], "thread-store-refused");
   return out;
 }
 
+// Row 269 (264b, #1774): one session's store. A legacy session (op legacy-3l,
+// whose id is its thread's) keeps the thread's store, so its harness resumes
+// the history it has; any other session gets its own under sessions/, with
+// the same checks on every segment and directory. Same shape as threadStore.
+// legacyThread: the thread has a legacy session. When it has none, a
+// threads/<b>/<t>/session left by a 264a run of the thread's default session
+// refuses legacy-store-unmapped, so that history isn't silently left behind.
+export function sessionStore(dataRoot, business, thread, session, { legacy = false, legacyThread = legacy } = {}) {
+  if (legacy) return threadStore(dataRoot, business, thread);
+  segments("thread store", [["business", business], ["thread", thread], ["session", session]], "thread-store-refused");
+  const threadDir = join(dataRoot, "threads", business, thread);
+  if (!legacyThread) {
+    let old = true;
+    try {
+      lstatSync(join(threadDir, "session"));
+    } catch (e) {
+      if (e.code !== "ENOENT") throw e;
+      old = false;
+    }
+    if (old) throw seatError(`thread store: ${join(threadDir, "session")} holds history no legacy session maps`, "legacy-store-unmapped");
+  }
+  const dir = join(threadDir, "sessions", session);
+  const out = { dir, session: join(dir, "session"), projects: join(dir, "claude-projects") };
+  privateDirs("thread store", [join(dataRoot, "threads"), join(dataRoot, "threads", business), threadDir, join(threadDir, "sessions"), dir, out.session, out.projects], "thread-store-refused");
+  return out;
+}
+
+const inside = (a, b) => a === b || a.startsWith(b + sep);
+const realOrNull = (p) => {
+  try {
+    return realpathSync(p);
+  } catch (e) {
+    if (e.code === "ENOENT") return null;
+    throw e;
+  }
+};
+// Row 269: a thread session's workspace, session-workspaces/<b>/<i>/<s>. Each
+// segment is an id and each directory ours, real and private
+// (session-workspace-unsafe). Then the realpaths: it may not be inside, or
+// contain, the legacy workspaces root, its business and instance directories
+// or any entry of either (old instance, .retired and task workspaces), nor
+// sit above or below a `live` workspace (session-workspace-overlap). The same
+// path as a live one is left to the lock, which refuses session-workspace-busy.
+export function sessionWorkspace(dataRoot, business, instance, session, { live = [] } = {}) {
+  segments("session workspace", [["business", business], ["instance", instance], ["session", session]], "session-workspace-unsafe");
+  const root = join(dataRoot, "session-workspaces");
+  const path = join(root, business, instance, session);
+  privateDirs("session workspace", [root, join(root, business), join(root, business, instance), path], "session-workspace-unsafe");
+  const real = realpathSync(path);
+  const legacy = join(dataRoot, "workspaces");
+  const named = [legacy, join(legacy, business), join(legacy, business, instance)];
+  const entries = (d) => {
+    try {
+      return readdirSync(d).map((x) => join(d, x));
+    } catch (e) {
+      if (e.code === "ENOENT" || e.code === "ENOTDIR") return [];
+      throw e;
+    }
+  };
+  for (const p of [...named, ...entries(legacy), ...entries(join(legacy, business))]) {
+    const other = realOrNull(p);
+    if (other && (inside(real, other) || inside(other, real))) throw seatError(`session workspace: ${path} overlaps the legacy workspace ${p}`, "session-workspace-overlap");
+  }
+  for (const p of live) {
+    const other = realOrNull(p);
+    if (other && other !== real && (inside(real, other) || inside(other, real))) throw seatError(`session workspace: ${path} overlaps the live workspace ${p}`, "session-workspace-overlap");
+  }
+  return path;
+}
+
+// Row 269: the single-writer lock on a session workspace, <workspace>.lock,
+// beside it and outside every mount. It names the host (pid, start time) and,
+// once spawned, the run's outer process. A lock whose host and run are both
+// gone is stale and is taken over; any other refuses session-workspace-busy.
+// Returns { file, update(fields), release() }.
+const lockAlive = (pid, start) => Number.isSafeInteger(pid) && pid > 0 && start != null && startTimeOf(pid) === start;
+export function lockSessionWorkspace(workspace, owner) {
+  const file = `${workspace}.lock`;
+  let held = { ...owner };
+  const write = (flag) => writeFileSync(file, `${JSON.stringify(held)}\n`, { mode: 0o600, flag });
+  const busy = (why) => seatError(`session workspace ${workspace} is busy: ${why}`, "session-workspace-busy");
+  for (let attempt = 0; attempt < 2; attempt++) {
+    try {
+      write("wx");
+      // One release per lock: a second unlink could take a lock another host has written since.
+      let released = false;
+      return {
+        file,
+        update(fields) {
+          if (released) return;
+          held = { ...held, ...fields };
+          write("w");
+        },
+        release() {
+          if (released) return;
+          released = true;
+          try {
+            unlinkSync(file);
+          } catch (e) {
+            if (e.code !== "ENOENT") throw e;
+          }
+        },
+      };
+    } catch (e) {
+      if (e.code !== "EEXIST") throw e;
+    }
+    let raw;
+    try {
+      raw = readFileSync(file, "utf8");
+    } catch (e) {
+      if (e.code === "ENOENT") continue;
+      throw e;
+    }
+    let other = null;
+    try {
+      other = JSON.parse(raw);
+    } catch {}
+    // An unreadable lock may be one being written, so it counts as held.
+    if (!other || typeof other !== "object") throw busy(`${file} is unreadable`);
+    if (lockAlive(other.hostPid, other.hostStartTime)) throw busy(`host pid ${other.hostPid} holds it`);
+    if (lockAlive(other.runPid, other.runStartTime)) throw busy(`run pid ${other.runPid} holds it`);
+    // Stale. Move it aside; if what moved isn't what was judged stale, another
+    // host took it over in between, so it goes back and this start refuses.
+    const aside = `${file}.stale-${process.pid}-${randomBytes(4).toString("hex")}`;
+    try {
+      renameSync(file, aside);
+    } catch (e) {
+      if (e.code === "ENOENT") continue;
+      throw e;
+    }
+    if (readFileSync(aside, "utf8") !== raw) {
+      try {
+        linkSync(aside, file);
+      } catch {}
+      unlinkSync(aside);
+      throw busy(`another host took ${file} over`);
+    }
+    unlinkSync(aside);
+  }
+  throw busy(`${file} could not be taken`);
+}
+
 // The session's environment is an allowlist: nothing else from the host's
 // environment passes, so a founder credential variable (GITEA_TOKEN,
 // GH_TOKEN, SSH_AUTH_SOCK, ...) never reaches a session (REQ-CRED-2).
@@ -212,10 +371,10 @@ export class Registry {
   count(business, provider, model = undefined) {
     return this.#sessions.filter((s) => s.business === business && s.provider === provider && (model === undefined || s.model === model)).length;
   }
-  // A thread instance runs once per thread (row 121), so `thread` is part of
-  // the key; null is a session that holds no thread.
-  running(business, instance, thread = null) {
-    return this.#sessions.find((s) => s.business === business && s.instance === instance && (s.thread ?? null) === thread) ?? null;
+  // A thread instance runs once per thread session (row 121, row 269), so
+  // `thread` and `session` are part of the key; null is an entry without one.
+  running(business, instance, thread = null, session = null) {
+    return this.#sessions.find((s) => s.business === business && s.instance === instance && (s.thread ?? null) === thread && (s.session ?? null) === session) ?? null;
   }
   #write() {
     if (this.#readOnly) return;
@@ -238,7 +397,7 @@ export function readSessions(dataRoot) {
   } catch {
     throw new SeatError(`sessions file is unreadable: ${file}`, 2);
   }
-  if (doc?.sessionsVersion !== SESSIONS_VERSION || !Array.isArray(doc.sessions)) throw new SeatError(`sessions file is malformed: ${file}`, 2);
+  if (!SESSIONS_READ.includes(doc?.sessionsVersion) || !Array.isArray(doc.sessions)) throw new SeatError(`sessions file is malformed: ${file}`, 2);
   return doc.sessions.map((s) => ({ ...s, live: startTimeOf(s.runnerPid) === s.runnerStartTime }));
 }
 

packages/seat/tests/compactions.test.mjs

@@ -0,0 +1,230 @@
+// Row 270 (264c, #1775, T1 item 7 D6, T3's Pi half): the compaction count
+// from a session's native Pi store, through the host ledger
+// (src/compactions.mjs). The streams are Pi 0.85.1's shapes, written by hand:
+// a session header, then entries, one JSON object per line.
+
+import { test } from "node:test";
+import assert from "node:assert/strict";
+import { appendFileSync, lstatSync, mkdirSync, mkdtempSync, readFileSync, rmSync, statSync, symlinkSync, unlinkSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join, relative, isAbsolute } from "node:path";
+import { collectCompactions, compactionCount, LEDGER, readLedger, storeDir } from "../src/compactions.mjs";
+import { binds } from "../src/sandbox.mjs";
+import { sessionStore } from "../src/session.mjs";
+
+function setup(t, { legacy = false } = {}) {
+  const root = mkdtempSync(join(tmpdir(), "seat-compactions-"));
+  t.after(() => rmSync(root, { recursive: true, force: true }));
+  const store = sessionStore(root, "acme", "th1", legacy ? "th1" : "s1", { legacy });
+  let tick = 0;
+  const now = () => new Date(Date.UTC(2026, 9, 22, 12, 0, tick++)).toISOString();
+  const collect = (harness = "pi") => collectCompactions(store, harness, { now });
+  const ledger = () => readFileSync(join(store.dir, LEDGER), "utf8");
+  const records = () => ledger().trim().split("\n").map((l) => JSON.parse(l));
+  return { root, store, collect, ledger, records };
+}
+
+const header = (id) => ({ type: "session", version: 3, id, timestamp: "2026-10-22T11:00:00.000Z", cwd: "/w" });
+const message = (id, role = "user") => ({ type: "message", id, parentId: null, timestamp: "2026-10-22T11:00:01.000Z", message: { role, content: "x" } });
+const compaction = (id, timestamp) => ({ type: "compaction", id, parentId: null, timestamp, summary: "s", firstKeptEntryId: "k", tokensBefore: 100 });
+const line = (e) => `${JSON.stringify(e)}\n`;
+const stream = (store, name, entries) => writeFileSync(join(store.session, name), entries.map(line).join(""));
+const append = (store, name, ...entries) => appendFileSync(join(store.session, name), entries.map(line).join(""));
+
+test("an empty store, a store with no compaction and a failed start count 0, known", (t) => {
+  const f = setup(t);
+  assert.deepEqual(f.collect(), { count: 0, known: true, reason: null, at: "2026-10-22T12:00:00.000Z" });
+  // A started and aborted compaction writes no entry; a compactionSummary message, a
+  // branch_summary and a message are not compactions.
+  stream(f.store, "a.jsonl", [header("a"), message("m1"), message("m2", "compactionSummary"), { type: "branch_summary", id: "b1", parentId: null, timestamp: "2026-10-22T11:00:02.000Z", summary: "b" }]);
+  assert.deepEqual(f.collect(), { count: 0, known: true, reason: null, at: "2026-10-22T12:00:01.000Z" });
+});
+
+test("one and three completed compactions", (t) => {
+  const f = setup(t);
+  stream(f.store, "a.jsonl", [header("a"), message("m1"), compaction("c1", "2026-10-22T11:01:00.000Z")]);
+  assert.equal(f.collect().count, 1);
+  append(f.store, "a.jsonl", message("m2"), compaction("c2", "2026-10-22T11:02:00.000Z"), compaction("c3", "2026-10-22T11:03:00.000Z"));
+  assert.deepEqual(f.collect(), { count: 3, known: true, reason: null, at: "2026-10-22T12:00:01.000Z" });
+  assert.equal(compactionCount(f.store.dir).count, 3);
+});
+
+// sage-2814: Pi's createBranchedSession and forkFrom copy entries into a new stream with
+// their id and timestamp, so a copied compaction counts once.
+test("a compaction copied into a branched stream counts once, under the stream it was first seen in", (t) => {
+  const f = setup(t);
+  const c1 = compaction("c1", "2026-10-22T11:01:00.000Z");
+  stream(f.store, "a.jsonl", [header("a"), message("m1"), c1]);
+  assert.equal(f.collect().count, 1);
+  stream(f.store, "b.jsonl", [header("b"), message("m1"), c1, message("m3")]);
+  assert.deepEqual(f.collect(), { count: 1, known: true, reason: null, at: "2026-10-22T12:00:01.000Z" });
+  append(f.store, "b.jsonl", compaction("c2", "2026-10-22T11:05:00.000Z"));
+  assert.equal(f.collect().count, 2);
+  const lines = f.records().filter((r) => r.kind === "compaction");
+  assert.deepEqual(lines.map((r) => [r.entry, r.stream]), [["c1", "a.jsonl"], ["c2", "b.jsonl"]]);
+});
+
+test("one collection over two streams holding the same copy counts it once and writes it once", (t) => {
+  const f = setup(t);
+  const c1 = compaction("c1", "2026-10-22T11:01:00.000Z");
+  stream(f.store, "a.jsonl", [header("a"), c1]);
+  stream(f.store, "b.jsonl", [header("b"), c1]);
+  assert.equal(f.collect().count, 1);
+  assert.equal(f.records().filter((r) => r.kind === "compaction").length, 1);
+});
+
+// Pi's ids are unique only within a file: the same id at another time is another compaction.
+test("the same entry id at another time in another stream is another compaction", (t) => {
+  const f = setup(t);
+  stream(f.store, "a.jsonl", [header("a"), compaction("c1", "2026-10-22T11:01:00.000Z")]);
+  stream(f.store, "b.jsonl", [header("b"), compaction("c1", "2026-10-22T11:09:00.000Z")]);
+  assert.equal(f.collect().count, 2);
+});
+
+test("resume: each collection reads past its cursor and writes a key once", (t) => {
+  const f = setup(t);
+  stream(f.store, "a.jsonl", [header("a"), compaction("c1", "2026-10-22T11:01:00.000Z")]);
+  f.collect();
+  f.collect();
+  append(f.store, "a.jsonl", compaction("c2", "2026-10-22T11:02:00.000Z"));
+  assert.equal(f.collect().count, 2);
+  const r = f.records();
+  assert.deepEqual(r.filter((x) => x.kind === "compaction").map((x) => x.entry), ["c1", "c2"]);
+  assert.deepEqual(r.filter((x) => x.kind === "cursor").map((x) => x.bytes), [statSync(join(f.store.session, "a.jsonl")).size - line(compaction("c2", "2026-10-22T11:02:00.000Z")).length, statSync(join(f.store.session, "a.jsonl")).size]);
+  assert.equal(r.filter((x) => x.kind === "collected").length, 3);
+  assert.equal(lstatSync(join(f.store.dir, LEDGER)).mode & 0o777, 0o600);
+});
+
+test("a partial trailing line waits for its newline", (t) => {
+  const f = setup(t);
+  stream(f.store, "a.jsonl", [header("a")]);
+  const whole = line(compaction("c1", "2026-10-22T11:01:00.000Z"));
+  appendFileSync(join(f.store.session, "a.jsonl"), whole.slice(0, 30));
+  assert.deepEqual(f.collect(), { count: 0, known: true, reason: null, at: "2026-10-22T12:00:00.000Z" });
+  appendFileSync(join(f.store.session, "a.jsonl"), whole.slice(30));
+  assert.deepEqual(f.collect(), { count: 1, known: true, reason: null, at: "2026-10-22T12:00:01.000Z" });
+});
+
+test("a missing ledger is not-collected and an unreadable one store-unreadable, never 0", (t) => {
+  const f = setup(t);
+  assert.deepEqual(compactionCount(f.store.dir), { count: null, known: false, reason: "not-collected", at: null });
+  mkdirSync(join(f.store.dir, LEDGER));
+  assert.deepEqual(compactionCount(f.store.dir), { count: null, known: false, reason: "store-unreadable", at: null });
+  assert.throws(() => f.collect());
+});
+
+test("a symlinked ledger is store-unreadable and is not written through", (t) => {
+  const f = setup(t);
+  const target = join(f.root, "elsewhere.jsonl");
+  writeFileSync(target, "");
+  symlinkSync(target, join(f.store.dir, LEDGER));
+  assert.equal(compactionCount(f.store.dir).reason, "store-unreadable");
+  assert.throws(() => f.collect());
+  assert.equal(readFileSync(target, "utf8"), "");
+});
+
+// sage-2806 condition 3: a torn last line is a bad line, not a skipped one.
+test("a torn last ledger line makes the count unknown, a lower bound, and stays so", (t) => {
+  const f = setup(t);
+  assert.equal(f.collect().count, 0);
+  appendFileSync(join(f.store.dir, LEDGER), '{"v":1,"kind":"compac');
+  assert.deepEqual(compactionCount(f.store.dir), { count: null, known: false, reason: "store-incomplete", at: "2026-10-22T12:00:00.000Z" });
+  stream(f.store, "a.jsonl", [header("a"), compaction("c1", "2026-10-22T11:01:00.000Z")]);
+  assert.deepEqual(f.collect(), { count: 1, known: false, reason: "store-incomplete", at: "2026-10-22T12:00:01.000Z" });
+  // The next write started on a fresh line: every line after the torn one is whole.
+  const lines = f.ledger().split("\n");
+  assert.equal(lines.pop(), "");
+  const bad = lines.filter((l) => {
+    try {
+      JSON.parse(l);
+      return false;
+    } catch {
+      return true;
+    }
+  });
+  assert.deepEqual(bad, ['{"v":1,"kind":"compac']);
+  assert.equal(readLedger(f.store.dir).events.size, 1);
+});
+
+test("a bad ledger line in the middle makes the count unknown", (t) => {
+  const f = setup(t);
+  stream(f.store, "a.jsonl", [header("a"), compaction("c1", "2026-10-22T11:01:00.000Z")]);
+  f.collect();
+  appendFileSync(join(f.store.dir, LEDGER), "not json\n");
+  append(f.store, "a.jsonl", compaction("c2", "2026-10-22T11:02:00.000Z"));
+  assert.deepEqual(f.collect(), { count: 2, known: false, reason: "store-incomplete", at: "2026-10-22T12:00:01.000Z" });
+});
+
+test("a stream that shrank, changed under its cursor or went away is store-incomplete for good", async (t) => {
+  for (const change of ["shrink", "rewrite", "remove"]) {
+    await t.test(change, (t) => {
+      const f = setup(t);
+      stream(f.store, "a.jsonl", [header("a"), compaction("c1", "2026-10-22T11:01:00.000Z")]);
+      assert.equal(f.collect().known, true);
+      const path = join(f.store.session, "a.jsonl");
+      if (change === "shrink") stream(f.store, "a.jsonl", [header("a")]);
+      // Same length, other bytes: only the cursor's tail digest sees it.
+      if (change === "rewrite") stream(f.store, "a.jsonl", [header("a"), compaction("c9", "2026-10-22T11:01:00.000Z")]);
+      if (change === "remove") unlinkSync(path);
+      const c = f.collect();
+      assert.equal(c.known, false);
+      assert.equal(c.reason, "store-incomplete");
+      assert.equal(c.count, change === "rewrite" ? 2 : 1);
+      // A later clean collection doesn't make it known again.
+      assert.equal(f.collect().reason, "store-incomplete");
+    });
+  }
+});
+
+test("a stream that isn't a regular file or a Pi session is store-unreadable, and that passes", (t) => {
+  const f = setup(t);
+  const target = join(f.root, "outside.jsonl");
+  writeFileSync(target, line(header("x")) + line(compaction("c1", "2026-10-22T11:01:00.000Z")));
+  symlinkSync(target, join(f.store.session, "link.jsonl"));
+  assert.deepEqual(f.collect(), { count: null, known: false, reason: "store-unreadable", at: "2026-10-22T12:00:00.000Z" });
+  unlinkSync(join(f.store.session, "link.jsonl"));
+  assert.deepEqual(f.collect(), { count: 0, known: true, reason: null, at: "2026-10-22T12:00:01.000Z" });
+});
+
+test("a first line that isn't a session header is store-unreadable", (t) => {
+  const f = setup(t);
+  stream(f.store, "a.jsonl", [compaction("c1", "2026-10-22T11:01:00.000Z")]);
+  assert.equal(f.collect().reason, "store-unreadable");
+});
+
+// RULINGS Q6: Claude Code is unknown with the Vikunja 466 reason, never 0, and a session any
+// Claude run collected for stays unknown.
+test("a Claude run is harness-unsupported, and stays so after a Pi run", (t) => {
+  const f = setup(t);
+  assert.deepEqual(f.collect("claude"), { count: null, known: false, reason: "harness-unsupported", at: "2026-10-22T12:00:00.000Z" });
+  stream(f.store, "a.jsonl", [header("a"), compaction("c1", "2026-10-22T11:01:00.000Z")]);
+  assert.deepEqual(f.collect("pi"), { count: 1, known: false, reason: "harness-unsupported", at: "2026-10-22T12:00:01.000Z" });
+});
+
+test("storeDir is the store sessionStore makes, without making it", (t) => {
+  const f = setup(t);
+  assert.equal(storeDir(f.root, "acme", "th1", "s1"), f.store.dir);
+  assert.equal(storeDir(f.root, "acme", "th1", "th1", { legacy: true }), join(f.root, "threads", "acme", "th1"));
+  assert.equal(storeDir(f.root, "acme", "th1", "s2"), join(f.store.dir, "..", "s2"));
+  assert.throws(() => lstatSync(join(f.store.dir, "..", "s2")), { code: "ENOENT" });
+  for (const bad of ["..", "a/b", "", "A", null]) assert.equal(storeDir(f.root, "acme", "th1", bad), null);
+  assert.equal(storeDir(f.root, "../x", "th1", "s1"), null);
+});
+
+// sage-2806 condition 4: the agent can't reach the ledger. It sits in the store dir, which no
+// bind mounts; only the store's session/ and claude-projects/ are.
+test("the ledger is under no sandbox mount", (t) => {
+  for (const legacy of [false, true]) {
+    const f = setup(t, { legacy });
+    f.collect();
+    const file = join(f.store.dir, LEDGER);
+    const runDir = join(f.root, "runs", "r1");
+    const list = binds({ runDir, workspace: join(f.root, "ws"), adapter: join(f.root, "a.sh"), brokerSocket: join(f.root, "bus.sock"), threadStore: f.store });
+    assert.ok(list.some((b) => b.src === f.store.session));
+    for (const b of list) {
+      if (!b.src || !isAbsolute(b.src)) continue;
+      const rel = relative(b.src, file);
+      assert.ok(rel.startsWith("..") || isAbsolute(rel), `${b.src} mounts ${file}`);
+    }
+  }
+});

packages/seat/tests/session-workspace.test.mjs

@@ -0,0 +1,126 @@
+// Row 269 (264b, #1774): a thread session's store, its workspace and the
+// single-writer lock beside the workspace (session.mjs sessionStore(),
+// sessionWorkspace(), lockSessionWorkspace()), and sessions.json v2.
+
+import { test } from "node:test";
+import assert from "node:assert/strict";
+import { spawn } from "node:child_process";
+import { existsSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, realpathSync, rmSync, statSync, symlinkSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { dirname, join } from "node:path";
+import { startTimeOf } from "../src/proc.mjs";
+import { SeatError } from "../src/seat.mjs";
+import { lockSessionWorkspace, readSessions, Registry, sessionsFile, sessionStore, sessionWorkspace, threadStore } from "../src/session.mjs";
+
+function scratch(t) {
+  const dir = realpathSync(mkdtempSync(join(tmpdir(), "mosaic-session-ws-")));
+  t.after(() => rmSync(dir, { recursive: true, force: true }));
+  return dir;
+}
+const mode = (p) => statSync(p).mode & 0o777;
+const refuses = (code) => (e) => e instanceof SeatError && e.code === code;
+
+test("REQ-THREAD-2 sessionStore: a legacy session keeps the thread's store, any other gets sessions/<s>/, each segment checked first", (t) => {
+  const root = scratch(t);
+  assert.deepEqual(sessionStore(root, "acme", "th-1", "th-1", { legacy: true }), threadStore(root, "acme", "th-1"));
+  const s = sessionStore(root, "acme", "th-1", "s-2", { legacyThread: true });
+  const dir = join(root, "threads", "acme", "th-1", "sessions", "s-2");
+  assert.deepEqual(s, { dir, session: join(dir, "session"), projects: join(dir, "claude-projects") });
+  for (const d of [dirname(dir), dir, s.session, s.projects]) assert.equal(mode(d), 0o700, d);
+  for (const [thread, session] of [["..", "s"], ["th-1", ".."], ["th-1", "a/b"], ["th-1", null]])
+    assert.throws(() => sessionStore(root, "acme", thread, session, { legacyThread: true }), refuses("thread-store-refused"), `${thread}/${session}`);
+  assert.deepEqual(readdirSync(join(root, "threads", "acme", "th-1", "sessions")), ["s-2"], "a refused call makes nothing");
+});
+
+test("REQ-THREAD-2 sessionStore refuses legacy-store-unmapped for a thread whose old store no legacy session maps", (t) => {
+  const root = scratch(t);
+  threadStore(root, "acme", "th-1");
+  assert.throws(() => sessionStore(root, "acme", "th-1", "s-1"), (e) => refuses("legacy-store-unmapped")(e) && /holds history no legacy session maps$/.test(e.message));
+  assert.equal(existsSync(join(root, "threads", "acme", "th-1", "sessions")), false);
+  // A thread with a legacy session: its other sessions take D2's layout.
+  assert.ok(sessionStore(root, "acme", "th-1", "s-1", { legacyThread: true }).dir.endsWith(join("sessions", "s-1")));
+  const elsewhere = join(root, "elsewhere");
+  mkdirSync(elsewhere, { mode: 0o700 });
+  symlinkSync(elsewhere, join(root, "threads", "acme", "th-2"));
+  assert.throws(() => sessionStore(root, "acme", "th-2", "s-1"), (e) => refuses("thread-store-refused")(e) && /is unsafe$/.test(e.message));
+  assert.deepEqual(readdirSync(elsewhere), []);
+});
+
+test("REQ-THREAD-2 sessionWorkspace: private directories per session, unsafe segments and overlaps refused", (t) => {
+  const root = scratch(t);
+  const ws = sessionWorkspace(root, "acme", "assistant", "s-1");
+  assert.equal(ws, join(root, "session-workspaces", "acme", "assistant", "s-1"));
+  for (const d of [join(root, "session-workspaces"), dirname(dirname(ws)), dirname(ws), ws]) assert.equal(mode(d), 0o700, d);
+  assert.equal(sessionWorkspace(root, "acme", "assistant", "s-1"), ws, "a second call reuses it");
+  assert.notEqual(sessionWorkspace(root, "acme", "assistant", "s-2"), ws);
+  for (const session of ["..", "a/b", ""]) assert.throws(() => sessionWorkspace(root, "acme", "assistant", session), refuses("session-workspace-unsafe"), session);
+  // A live workspace at the same path is the lock's to refuse; one above or below it overlaps.
+  assert.equal(sessionWorkspace(root, "acme", "assistant", "s-1", { live: [ws] }), ws);
+  assert.throws(() => sessionWorkspace(root, "acme", "assistant", "s-3", { live: [dirname(ws)] }), refuses("session-workspace-overlap"));
+  // The legacy root symlinked into a session path: every segment is real, only the realpath finds it.
+  mkdirSync(join(root, "workspaces"), { mode: 0o700 });
+  symlinkSync(ws, join(root, "workspaces", "acme"));
+  assert.throws(() => sessionWorkspace(root, "acme", "assistant", "s-1"), refuses("session-workspace-overlap"));
+  // A symlinked root fails the segment check first.
+  const other = scratch(t);
+  const elsewhere = join(other, "elsewhere");
+  mkdirSync(elsewhere, { mode: 0o700 });
+  symlinkSync(elsewhere, join(other, "session-workspaces"));
+  assert.throws(() => sessionWorkspace(other, "acme", "assistant", "s-1"), refuses("session-workspace-unsafe"));
+  assert.deepEqual(readdirSync(elsewhere), []);
+});
+
+test("REQ-THREAD-2 lockSessionWorkspace: one writer; a live host or run holds it, a stale lock is taken over, an unreadable one holds", async (t) => {
+  const root = scratch(t);
+  const ws = sessionWorkspace(root, "acme", "assistant", "s-1");
+  const me = { hostPid: process.pid, hostStartTime: startTimeOf(process.pid) };
+  const lock = lockSessionWorkspace(ws, me);
+  assert.equal(lock.file, `${ws}.lock`);
+  assert.equal(mode(lock.file), 0o600);
+  assert.throws(() => lockSessionWorkspace(ws, me), refuses("session-workspace-busy"));
+  lock.update({ run: "r-1", runPid: 1, runStartTime: "x" });
+  assert.deepEqual(JSON.parse(readFileSync(lock.file, "utf8")), { ...me, run: "r-1", runPid: 1, runStartTime: "x" });
+  lock.release();
+  assert.equal(existsSync(lock.file), false);
+
+  // A run that outlives its host still holds it.
+  const run = spawn("sleep", ["60"], { stdio: "ignore" });
+  const gone = new Promise((r) => run.once("exit", r));
+  t.after(() => run.kill("SIGKILL"));
+  let runStart = null;
+  for (let i = 0; i < 100 && !runStart; i++) {
+    runStart = startTimeOf(run.pid);
+    if (!runStart) await new Promise((r) => setTimeout(r, 20));
+  }
+  const held = `${JSON.stringify({ hostPid: 2 ** 22 + 7, hostStartTime: "gone", runPid: run.pid, runStartTime: runStart })}\n`;
+  writeFileSync(`${ws}.lock`, held, { mode: 0o600 });
+  assert.throws(() => lockSessionWorkspace(ws, me), refuses("session-workspace-busy"));
+  assert.equal(readFileSync(`${ws}.lock`, "utf8"), held);
+  run.kill("SIGKILL");
+  await gone;
+  // Now stale: taken over, and nothing is left beside it.
+  const taken = lockSessionWorkspace(ws, me);
+  assert.deepEqual(JSON.parse(readFileSync(taken.file, "utf8")), me);
+  assert.deepEqual(readdirSync(dirname(ws)).sort(), ["s-1", "s-1.lock"]);
+  // One release per lock: a second doesn't remove a lock written since.
+  taken.release();
+  writeFileSync(`${ws}.lock`, "{}\n", { mode: 0o600 });
+  taken.release();
+  assert.equal(readFileSync(`${ws}.lock`, "utf8"), "{}\n");
+  // An unreadable lock may be one being written, so it holds.
+  writeFileSync(`${ws}.lock`, "{", { mode: 0o600 });
+  assert.throws(() => lockSessionWorkspace(ws, me), refuses("session-workspace-busy"));
+});
+
+test("REQ-THREAD-2 sessions.json v2: entries carry session and workspace, running() keys on the session, a v1 file still reads", (t) => {
+  const root = scratch(t);
+  const registry = new Registry(root);
+  const entry = { business: "acme", instance: "assistant", thread: "th-1", session: "s-1", workspace: "/w", run: "r-1" };
+  registry.add(entry);
+  assert.equal(JSON.parse(readFileSync(sessionsFile(root), "utf8")).sessionsVersion, 2);
+  assert.deepEqual(registry.running("acme", "assistant", "th-1", "s-1"), entry);
+  assert.equal(registry.running("acme", "assistant", "th-1", "s-2"), null);
+  assert.equal(registry.running("acme", "assistant", "th-1"), null, "an entry with a session isn't the thread's sessionless run");
+  writeFileSync(sessionsFile(root), `${JSON.stringify({ sessionsVersion: 1, sessions: [{ business: "acme", instance: "assistant", thread: "th-1", run: "r-0" }] })}\n`);
+  assert.deepEqual(readSessions(root).map((s) => [s.run, s.session ?? null]), [["r-0", null]]);
+});