Compare commits

..
34 changed files with 1703 additions and 2610 deletions

No files matched your search

+170
View File
@@ -0,0 +1,170 @@
# Row 224 — C2-05b broker lookups and typed tools (#1725)
Owner: Della; reviewer: Dewey. Assignment sage-2795; scope extension sage-2804.
Base: `4f3fbc25a` (Researcher C2-05a). Branch: `della/row224`.
Brief: C2-05b in `docs/plans/2026-10-20_case2-rows.md` at `aca16404c`
and `51cc0445c`; design section 3, 5.3 M3 and 7.2; C2-05a BUILD rulings.
## Execution boundary and progress
Code/docs and local Git only while row 253 holds the gate. No Node (including
syntax checks), HTTPS Git, queue claim, test or push until Sage's CLEAR.
Tests are written first; their RED/GREEN runs are deferred, not claimed.
Queue next is deferred too: its executable uses Node. Sage assigned this row
explicitly. After 253 and 216 land, rebase under a new branch name.
- Pre-flight: query.mjs consumes `{repo,commit,mapping}`; broker cannot import
Git/filesystem library code because it is in the sandbox import closure.
Runtime injects the host lookup adapter; existing human file/status/sync stay.
- Pre-flight: resolved has no project list and a library is never a checkout.
Bundle needs an explicit library flag from launcher business configuration.
- M3 reading: exact creative base is five lookups plus send_message; retain
explicitly granted broker web_search/web_fetch for research/companion under
landed C2-08 and sage-2804. No generic spec, task, evidence or decision tools. Residual (mos-1308): those
reads lack broker action checks; their typed-tool absence is the only fence
for creative roles. M1/M2 deny bash; broker-side refusal is outside this row.
Mos confirmed this interpretation in mos-1308; design wording fix belongs to Gosalyn. Other definitions retain their current tools.
- Multi-library businesses use an optional project ID; omit it only with one
configured library. Universe is required for all five tools, including get.
- Missing library uses the brief's `library-not-configured`; argument errors
reuse `invalid-request` and C2-05a query codes. No new action or authority.
## Manifest (draft code, unrun)
- bus/src/broker.mjs: five lookup verbs, strict args, holder/business rechecks.
- bus/src/runtime.mjs: host lookup adapter, stale-before-read sync, served pin.
- harness/src/tools.mjs: five schemas; creative exact-set filtering.
- harness/src/bundle.mjs: library flag forwarded to typedTools.
- cli/src/launcher.mjs: one flag derived from trusted business projects.
- bus/tests/library-lookups.test.mjs: real query/broker/socket/runtime cases.
- harness/tests/library-tools.test.mjs: schemas, role snapshots, bundle parity.
- cli/tests/thread-launch.test.mjs: flag reaches a real launched bundle with no library checkout.
- agents/della/work/queue-224/BUILD.md: manifest, rulings and deferred verification ledger.
## Verification ledger
No test or syntax-check run yet. Acceptance requires full bus/cli/harness suites
and every scripts/test-*.sh on Sage's gate; focused files and affected CLI/
seat import-closure tests precede that only in an authorized CLEAR.
## Coordination
Huey272 reports no runtime/bundle/launcher overlap. Rocko219 changes launcher
THREAD_DEFINITIONS checks only, separate from our buildBundle flag argument.
The launcher argument is required: bundle cannot derive library presence from
checkout IDs (a library has no agent clone). No role, authority or config schema
changes; library project entries are already validated and protected by M9.
## Mount and pin impact (inspection only)
No new imports in broker.mjs or tools.mjs; no RUNNER_FILES or sandbox mount
addition. runtime.mjs is host-only and already imports library index.mjs.
Bundle forwards only a boolean, never a root/mapping/slot/credential. The
existing import-closure tests are bus/tests/project-runtime.test.mjs and
seat/tests/sandbox.test.mjs. Grep found no literal sha256 pin for runtime.mjs
or bundle.mjs; bundle tests derive current source digests through SOURCES.
Library sources remain in the existing whole-directory release copy.
## Runs to schedule only after CLEAR
Set TMPDIR=/mnt/storage/scratch/tmp. Focused new bus/harness files first,
then existing library.test.mjs, library-sync.test.mjs, project-runtime.test.mjs,
harness creative/tools/spec-tools/bundle files and affected CLI launch tests.
Follow with the required whole bus/cli/harness suites and Sage's full shell gate.
No timing claim yet: use the repository/Sage per-file limits and record every
command, outcome, elapsed time and timeout before acceptance. No focused skip
or pattern run substitutes for whole-file gate evidence.
## Read-through findings and residuals
- Library get requires universe although the older design table omitted it;
section 3 isolation rule and C2-05a's accepted API are binding.
- Query type enum matches mapping.TYPES (five values); manuscript is a distinct
lookup and not a find type. Unknown host-path/commit/business/accept fields
are refused; explicit project is restricted to this business's library list.
- Async reads recheck the live holder and current project revision. A stale
read attempts sync, then takes the new served snapshot; a returned fetch
failure/divergence keeps serving the old pin. Busy sync uses the current pin.
Unreadable state, no served commit or unsafe configuration refuse. Query
and sync never hold a SQLite transaction over await.
- Processing halt refuses new agent lookup work and rechecks before returning
async results. Existing human library file/status/sync behavior is unchanged.
- Timer sync and reads still use C2-03a's 60-second contract. No timer tuning or
resource-limit schema is introduced by this row. Q222 creative reply settings
are in C2-06; no reply guard or writing capability is implemented here.
- The generic tool socket's existing 60-second client timeout can precede the
server's 120-second Git timeout. This row does not change the generic timeout;
a timed-out lookup is read-only and retryable, never a write/outcome claim.
- All tests are authored but unrun. No RED, GREEN, syntax, runtime performance,
sandbox, broker/harness suite, shell gate, claim or push evidence exists yet.
## Candidate review checklist (read-through only)
- [x] Five verbs mapped to real C2-05a methods, all require universe.
- [x] Default sole library or explicit owned project; no caller repo/commit.
- [x] Human-only file/status/sync gates kept; lookup holder rechecked.
- [x] Async socket dispatch includes lookup verbs through LIBRARY_VERBS.
- [x] Stale/unknown fetch syncs first; results use current immutable served pin.
- [x] Six/eight creative names per mos-1308; other roles retain their tools.
- [x] Launcher -> bundle -> typedTools passes presence without a library clone.
- [x] No imports added to sandbox source closure; no mount change.
- [ ] Execute new tests, RED characterization, full files/suites after CLEAR.
- [ ] Rebase after rows 253/216 land, under a new branch name.
- [ ] Claim/push only in Sage-authorized gap; Dewey review and Sage gate.
Candidate preservation commits are explicitly unrun under sage-2795; they are
not completion, green-suite or landing claims. No dependencies installed.
Correction: the first committed-range diff --check found a blank line at EOF
in the new library-tools test, missed while it was untracked. Removed in a
follow-up local commit; the complete range must be checked again. No tests ran.
## Shared-log receipts for Sage (sage-2815)
The shared logs are unchanged from the base in this row. Sage appends these
receipts in the landing sweep; they are preserved here, not in the manifest.
For `BUILD-LOG.md`:
```text
2026-10-22T16:36:30Z | Della row224 C2-05b #1725 preparation | Before: C2-05a queries exist at 4f3fbc25a but no broker lookup verbs or typed tools. Scope: broker/tools plus Sage-authorized runtime/bundle integration, tests and docs. Existing human-only library reads retained; five agent queries require a live holder and trusted configured source. Code/docs only under row253 gate; no Node/test/push/claim. Runtime ownership coordinated with Rocko219/Huey272.
```
For `docs/SESSIONS.md`:
```text
2026-10-22T16:36:30Z | Della (sage-2795 / sage-2804, Q220) | Row 224 C2-05b #1725, della/row224 base 4f3fbc25a | Read both pinned briefs and C2-05a source; preparing broker/typed tools and authorized host runtime/bundle seams. No queue command, Node, HTTPS Git, claim, test or push during Gander gate. Dewey reviewer; Rocko/Huey runtime coordination sent. Tests and integration verification deferred to CLEAR.
```
Local-only draft pins 19c0a7adf/7cf39c390 are superseded by the
log-free candidate. Saved as local branch della/row224-draft-with-logs before
rebuilding the unpublished candidate commit; no remote history changed.
## Dewey dewey-0852 and Sage sage-2819 — focused corrections
B1: inline freshness uses max(lastFetch, lastError.at), not lastFetch alone.
An error within 60 seconds counts as a recent attempt, so later lookups serve
the old served pin without another fetch. Keep inline sync: the brief requires
it and businesses without timed libraries have no sync timer. The extended
failed-fetch test makes reads at +1, +59,999 and +60,000 ms (one attempt total),
then +60,001 ms (second attempt). All reads must return the old commit.
Required mutant M-B1: remove lastError.at from recentAttempt. The repeated
reads must fail the attempted-count assertion. This mutant has NOT been run.
Residual (Sage will defer): the first inline fetch in each outage window can
still exceed the tool timeout (60 seconds, or the runner broker client’s
30 seconds). Bounding that fetch is out of scope for 224. Subsequent reads
are paced by persisted lastError.at. No claim of first-read outage latency.
N1: whole bus, cli and harness suites are required, plus the shell gate.
If the new thread-launch integration test skips due to sandboxProblem(),
name the skip; a skip is not launcher integration proof.
N2: cli/tests/thread-launch.test.mjs now depends on library/tests/helpers.mjs
mappingEntry() and library/tests/fixtures/mapping.json. Changes to that fixture
must rerun the CLI launch file.
N3: harness/tests/library-tools.test.mjs compares the type enum to mapping.mjs
TYPES, so adding a mapped type cannot silently leave the typed tool behind.
These edits are local code/docs only. No Node, RED/GREEN run, mutant, claim,
HTTPS Git or push. New local pin goes to Dewey/Mos/Sage for focused review.
-449
View File
@@ -1,449 +0,0 @@
# 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 |
-156
View File
@@ -1,156 +0,0 @@
# 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.
-417
View File
@@ -1,417 +0,0 @@
# 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
+184
View File
@@ -0,0 +1,184 @@
# Row 216 build notes (C2-05a, lookup index and queries, #1717)
Phase A only: code, tests and this file. No test, suite or mutant has been run. `node --check` passed on every changed `.mjs`.
- Branch: `researcher/row216`
- Base: `d372102882d6858fa77af4aee453f866c886ad5c` (origin/refactor)
- Design: `agents/gosalyn/work/case2/DESIGN.md` on origin/gosalyn/case2-design, section 3 (table at lines 625-631, bullets 633-641), 1.2-1.4, B12, D2
- Brief: `docs/plans/2026-10-20_case2-rows.md`, C2-05a at lines 320-357
## Files
| Path | Status |
|---|---|
| `packages/library/src/index-build.mjs` | new |
| `packages/library/src/query.mjs` | new |
| `packages/library/src/index.mjs` | edit: 2 export lines added |
| `packages/library/tests/query-fixture.mjs` | new, test helper |
| `packages/library/tests/query.test.mjs` | new |
| `packages/library/tests/index-build.test.mjs` | new |
| `agents/researcher/work/queue-216/BUILD.md` | new (this file) |
No existing test or source file changed beyond `index.mjs`.
## Design rows and where each is tested
| Design row | Code | Tests (query.test.mjs unless noted) |
|---|---|---|
| `library_find` | `find` | find is bound to its universe; every result carries its scope; long list pages; large items page under the cap |
| `library_get` | `get` | hybrid gets a source file; get lists visible links both ways; get pages its lines |
| `library_search` | `search` | search returns scoped line hits; item that cannot fit alone is refused |
| `library_manuscript` | `manuscript` | one chapter at most 400 lines; refused without chapter; hybrid reads a source chapter |
| `library_universe` | `universeInfo` | universe info reports a hybrid's sources, constraints, notes; unknown universe refused |
| Hybrid semantics (1.4) | `sees` | a hybrid sees its sources canon minus exclusions, no playground |
| Unmapped never appears | index build + `sees` | unmapped files never appear (query.test); index holds mapped files only (index-build.test) |
| Per-commit cache (3) | `indexFor` | same commit, path, mapping return one index; different path, mapping or commit get their own; eight-entry bound; failed read not kept; unchecked mapping refused (index-build.test) |
| Rebuild time (3, C2-05a gate) | `indexFor` | 3,000-file rebuild, time reported with `t.diagnostic`, no threshold asserted |
| REPLY_CAP parity | `QUERY_REPLY_CAP` | equals `REPLY_CAP` from harness tools.mjs |
## Choices where the design table is silent or differs
These are departures or gaps. Each is built as stated, so Sage can rule before Phase B. A ruling that changes one of them is a small edit.
1. **Paging arguments (gap a).** The table has no offset. `find` and `search` take an optional `offset` (a position in the filtered list, default 0). A reply carries `next`: the position of the next item, or `null`.
2. **`library_get` has no universe (gap b).** Built with a required `universe`. Without it the reply could not be isolated. A missing file and a file the universe cannot see both return `null`, so a caller cannot tell them apart. `get` also takes an optional `from` line.
3. **`get` over the cap (gap c).** Lines page by `from` with `next`. `links_out` and `links_in` come back whole on every page. If the frontmatter and links alone exceed the cap, the call is refused with `query-item-too-large`. A hub note with many backlinks may be refused. This needs a ruling: link paging, or accept the refusal.
4. **`manuscript` without a chapter (gap d).** Refused with `query-chapter-required`. The design names no list form. A chapter name that matches two visible files is refused with `query-chapter-ambiguous`.
5. **`search` excerpt (gap e).** The excerpt is the whole line, not a truncation. A line that alone exceeds the cap is refused. Truncating it would cut an item.
6. **`exclude` form (gap f).** `checkMapping` accepts only a folder path for `exclude` (`folder()`), so an exclusion is a path prefix. There is no type or source form to build. Confirm.
7. **Pin constraints.** `pin` does not hide any file. It is reported by `universeInfo` only. Confirm.
8. **Playground through `get`.** `scope` applies to lists. `get` by an explicit path returns a playground file of the caller's own universe. Confirm.
9. **Links.** A link into a file the universe cannot see is dropped from `links_out` and `links_in`, in line with N1. Links from a visible file that point into another universe are kept in the index (the reader keeps them under R6), and the query layer drops them for universes that cannot see them.
10. **`find` matching.** Case-insensitive substring over the front matter `name` and the file name. An empty `query` matches every file in scope.
11. **Cache key.** `repo`, `commit` and a sha256 of the canonical mapping. Keying by commit alone would return the wrong scopes after a mapping change. The cache holds eight entries, least recently used out. Failed reads are not kept.
12. **`universeInfo`.** Reads only the checked mapping and the served commit. It builds no index.
13. **Reply commit.** Every reply carries the index commit, which is the served commit passed in.
## Provisional refusal codes (pending Sage approval)
The brief says new refusal codes need Sage's approval first. These are provisional and must not be treated as settled:
- `query-argument-invalid`
- `query-universe-unknown`
- `query-chapter-required`
- `query-chapter-ambiguous`
- `query-item-too-large`
Reused, existing codes: `library-mapping-invalid`, plus the reader's `library-git`, `library-commit` through `indexFor`.
## Grep results (grep only, not a test run)
- Test references to the changed paths: `packages/library/tests/*.mjs` import `../src/index.mjs` and `../src/mapping.mjs`. `packages/bus/tests/library.test.mjs` and `library-sync.test.mjs` import the library index as a namespace. Neither enumerates its exports, so the two new export lines break no assertion.
- `packages/cli/tests/stack-copy.test.mjs:52` checks that `packages/library/src/index.mjs` exists, and it still does.
- 64-hex pins: none in `packages/library/tests`, none on `index.mjs`, `query.mjs` or `index-build.mjs`.
- `scripts/test-*.sh`: no reference to the new modules.
## Flag for Sage
- `packages/library/tests/helpers.mjs` `repo()` builds its fixtures under `tmpdir()`, which is `/tmp`. The launcher rules forbid that. The new tests do not use it; they use `tests/query-fixture.mjs`, which builds under `/mnt/storage/scratch/tmp`. The existing helper should be fixed in a separate row, not here.
## Phase B commands (not run)
Run only after Sage sends "slot clear". Set the scratch TMPDIR first, per the launcher rule.
```
cd /mnt/storage/src/mosaic-stack/.claude/worktrees/agent-aa3cc98787bc5152c/packages/library
export TMPDIR=/mnt/storage/scratch/tmp
node --test tests/query.test.mjs tests/index-build.test.mjs
node --test tests/*.test.mjs
```
Also run `packages/bus/tests/library.test.mjs` and `library-sync.test.mjs`, since they import the index.
## Status
Not yet verified. Nothing has been tested and nothing is committed. The working tree holds the Phase A changes. AGENTS.md invariant 8 allows a commit only after the applicable suites are green, so the commit waits for Phase B. The dispatch asked for a local commit in Phase A; Sage to confirm that the commit waits for Phase B.
## Round 2 (2026-10-22, sage-2708, sage-2716): rulings applied, rebased onto aca16404c
Added after the text above; nothing above is edited. "Nothing is committed", "Provisional refusal codes",
`query-chapter-required` and the tmpdir flag above were true when written and are superseded here.
- Branch `researcher/row216-r2`, one local commit on origin/refactor `aca16404c`. Row 268 (C2-01c) is in
that base. Row 253 has not landed; the work rebases onto 253's tip before any gate. Not pushed, not run.
- Rulings: `agents/sage/work/queue-216/RULINGS.md` at da9555ca2, rulings 1-13 and the tmpdir flag.
- Drift check, d37210288..aca16404c against my paths: the library files that changed are `mapping.mjs`,
`scope.mjs`, `README.md` and the mapping and scope tests. None is a file I add or edit. `query.mjs` reads
the checked mapping through them, so the whole library package runs in phase B.
### What changed from phase A
| Ruling | Change in `query.mjs` | Test |
|---|---|---|
| 3 `get` over the cap | `links_out` and `links_in` page together with `links_from`; every reply carries `links_next`, `next` and the front matter. A page with lines left carries at least one line. Links take at most half the cap while lines remain. `query-item-too-large` is left for front matter alone or one line alone (or one link alone) | hub note with 900 backlinks pages and adds up; front matter alone and a line alone refused; bad `links_from` |
| 4 `manuscript` no chapter | `chapter` optional: the first visible chapter in path order. Every reply carries `next_chapter`. `query-chapter-required` removed. Ambiguity still refuses | first chapter and `next_chapter`; ambiguous name refused |
| 5 `search` long line | hit comes back `{path, line, excerpt: null, excerpt_too_long: true, scope}`, other hits still return | marked hit and unmarked hits in one reply |
| 6 `exclude` | path-prefix reading only. The test names `EXCLUSION` and `EXCLUDED_PATHS` in `query-fixture.mjs`, so C2-15 (row 267) swaps two constants | exclusion hides from the hybrid and not from the source |
| 8 hybrid `get` | unchanged code; test added | hybrid gets a source's playground file as `null`; the source gets it with `scope: "playground"` |
| PRD 1.14 (row 268) | `universeInfo` reports `kind` from the mapping (`draft` included) | four tests below |
PRD 1.14 tests, one each: a lookup naming a `book` leaves `book: null` files out (find and search);
`library_manuscript` never reads a `book: null` file; a draft universe has no canon, so its lookups return
files only with `scope: "playground"`; `library_universe` on a draft returns kind, name and prefixes.
The levels fixture is `WITH_LEVELS` in `query.test.mjs`: harbor gets `Notes/Harbor/` as a universe-level
playground prefix, and `idea` is a draft over `Drafts/New World/`.
`query-fixture.mjs` builds under `tmpdir()`, as `helpers.mjs` does (ruling "tmpdir flag"). The earlier
flag about `helpers.mjs` building under `/tmp` is closed by that ruling: the gate sets `TMPDIR`.
Refusal codes now: `query-argument-invalid`, `query-universe-unknown`, `query-chapter-ambiguous`,
`query-item-too-large`. All four approved in the rulings.
### Grep results (grep only, not a test run)
- `git grep` over `packages/*/tests/*` and `scripts/*` for `query.mjs` and `index-build`: no hit outside
`packages/library/tests/query*.test.mjs` and `index-build.test.mjs`.
- 64-hex pins naming a library path: none. Positive control: the same pathspec finds the sha256 lines in
`packages/cli/tests/case1-hostcheck.test.mjs` and the `packages/library/src/index.mjs` existence check at
`packages/cli/tests/stack-copy.test.mjs:52`.
- Tests that read `packages/library/src` as a directory: `packages/cli/tests/stack-release-fixture.mjs:309`
copies `packages/library/src` whole into a release fixture. It carries no file list, so two new files break
no assertion. `stack-release.test.mjs` is not in the gate unless Sage wants it.
- Tests that import the library index as a namespace: `packages/bus/tests/library.test.mjs` and
`library-sync.test.mjs`. Neither enumerates exports, so the two export lines add no break. Both run
whole in phase B.
### Run limits (fixed before any run)
Every node file runs as its own `setsid -w timeout -k 10 LIMIT node --test FILE`; nothing runs unbounded.
- Node test files: LIMIT 900 s. The library test files have no recorded time. 900 s is above the longest
recorded whole-package run I know of (cli, 591 s, row 281 BUILD.md "Runs" step 3), so it holds the 3x rule
for any single file. The first run records each file's time, and a file past 300 s gets a limit re-derived
from it before the next run. `query.test.mjs` builds a 900-file repo and a 3,000-file index rebuild is in
`index-build.test.mjs`; both are the files most likely to need it.
- No shell script is in this row's gate, so no script limit applies.
Any timeout goes to Sage, then Mos, unplaced (mos-1275).
### Phase B (not run; Sage's CLEAR first, then a rebase onto 253's tip)
```
export TMPDIR=/mnt/storage/scratch/tmp
cd <worktree>/packages/library
for f in tests/*.test.mjs; do setsid -w timeout -k 10 900 node --test "$f"; done
cd ../bus
for f in tests/library.test.mjs tests/library-sync.test.mjs; do setsid -w timeout -k 10 900 node --test "$f"; done
cd ../cli
setsid -w timeout -k 10 900 node --test tests/stack-copy.test.mjs
```
Mutants I propose (my list, not read off the brief; each is meant to be killed by the named test, and none
has been thrown; the kill claims are unverified until CLEAR):
- M-a: `sees()` returns true for a file of another universe: killed by "find is bound to its universe".
- M-b: the playground filter dropped: killed by "playground appears only when asked for".
- M-c: the hybrid reads a source's playground: killed by "a hybrid gets a source playground file as null".
- M-d: `links_next` always null: killed by the hub paging test.
- M-e: the book filter lets `book: null` through: killed by the PRD 1.14 book test.
### Status
Built, not run. `node --check` has not been run on this round's files (no node before the CLEAR); the
phase A pass above covered the earlier text only. The commit is local and waits for no gate under Sage's
sage-2716 instruction, which allows git and docs work only; push and claim wait for Sage's go.
-2
View File
@@ -478,8 +478,6 @@ 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
+1 -1
View File
@@ -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,compactions}]`. `compactions` is `{count,known,reason,at}` from the host's ledger (row 270, #1775): `known` true gives the exact count of completed Pi compactions; otherwise `count` is a lower bound or null and `reason` is `harness-unsupported`, `not-collected`, `store-unreadable` or `store-incomplete`. Writes nothing. |
| `session.list` | `{thread}`; human and tester sessions (row 264a, #1769). The thread's sessions oldest first, `[{id,at,name,predecessor,default}]`. 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)
-37
View File
@@ -478,40 +478,3 @@ 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;
+58 -35
View File
@@ -96,7 +96,16 @@ export const TASK_PAGE_MAX = 200;
// request allowed past the ordinary 64 KiB envelope (server.mjs and client.mjs), up to its content cap.
export const PROJECT_FILE_VERBS = Object.freeze(['project.files', 'project.file', 'project.file.save']);
// C2-03a: library.sync is the human's fetch, and with `accept` the one way past library-diverged.
export const LIBRARY_VERBS = Object.freeze(['library.file', 'library.status', 'library.sync']);
// Row 224: pure agent lookups are separate from the three human-only verbs. Listing them
// here also makes server.mjs await their host effects instead of serializing a Promise as {}.
const LIBRARY_LOOKUPS = Object.freeze({
'library.find': { method: 'find', fields: ['universe', 'query', 'type', 'book', 'scope', 'offset'], required: ['universe', 'query'] },
'library.get': { method: 'get', fields: ['universe', 'path', 'from', 'links_from'], required: ['universe', 'path'] },
'library.search': { method: 'search', fields: ['universe', 'text', 'book', 'scope', 'offset'], required: ['universe', 'text'] },
'library.manuscript': { method: 'manuscript', fields: ['universe', 'book', 'scope', 'chapter', 'from', 'to'], required: ['universe', 'book'] },
'library.universe': { method: 'universeInfo', fields: ['universe'], required: ['universe'] },
});
export const LIBRARY_VERBS = Object.freeze(['library.file', 'library.status', 'library.sync', ...Object.keys(LIBRARY_LOOKUPS)]);
export const PROJECT_SAVE_VERB = 'project.file.save';
export const PROJECT_SAVE_REQUEST_MAX = 1024 * 1024 + 64 * 1024;
export const WEB_VERBS =Object.freeze(['web.search', 'web.fetch']);
@@ -384,7 +393,6 @@ 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.
@@ -410,14 +418,13 @@ 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, compactions = 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 }) {
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.
@@ -437,7 +444,6 @@ export class Broker {
this.#giteaGrant = giteaGrant;
this.#giteaKeys = giteaKeys;
this.#access = access;
this.#compactions = compactions;
this.#project = project;
this.#library = library;
this.#usage = usage;
@@ -755,15 +761,6 @@ 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');
@@ -849,7 +846,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 session of each thread (row 269, 264b): messages to the instance that no
// Trusted launcher API, not a socket verb. Per thread: 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
@@ -869,31 +866,30 @@ 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 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);
// 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));
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 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.
// 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.
#threadRows(where, ...params) {
return this.#store
.all(
`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
`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
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=se.id AND m.to_role=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
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=se.id) AS last_message,
(SELECT max(at) FROM thread_opens WHERE session=se.id) AS last_open,
(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,
c.at AS claim_at, c.op AS claim_op, c.holder_run AS holder_run
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`,
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`,
...params,
)
.map((r) => ({
@@ -905,8 +901,6 @@ 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.
@@ -4616,10 +4610,42 @@ export class Broker {
throw this.#refused(s, ProjectFailure && e instanceof ProjectFailure ? new BusError(e.code) : e, request);
}
}
// Row 224 C2-05b: the agent never supplies a host repo, commit or business. Its current
// holder and business entry are rechecked before/after awaits; runtime owns Git and queries.
async #requestLibraryLookup(s, { verb, args }) {
try {
this.#agent(s);
const business = this.#business(s.business);
const projects = Object.entries(business.projects ?? {}).filter(([, entry]) => entry.kind === 'library');
if (!projects.length || typeof this.#library?.lookup !== 'function') fail('library-not-configured');
const spec = LIBRARY_LOOKUPS[verb];
keys(args, ['project', ...spec.fields], spec.required);
if (!text(args.universe, 160)) fail('invalid-request');
if (args.project === undefined && projects.length !== 1) fail('invalid-request');
const project = args.project ?? projects[0][0];
if (!text(project, 160) || !projects.some(([name]) => name === project)) fail('library-not-configured');
const revision = JSON.stringify(business.projects[project]);
const authorize = () => {
this.#agent(s);
this.#processingGate(s, verb);
if (JSON.stringify(this.#business(s.business).projects?.[project]) !== revision) fail('library-not-configured');
};
const { project: ignored, ...query } = args;
authorize();
const result = await this.#library.lookup({ business, project }, { method: spec.method, args: query, authorize });
authorize();
this.#secretCheck(result);
return result;
} catch (error) {
const Failure = this.#library?.LibraryError;
throw this.#refused(s, Failure && error instanceof Failure ? new BusError(error.code) : error, { verb, args });
}
}
// C2-04 M4: only a verified human reads library text or status. Host Git
// functions stay injected, preserving this module's sandbox import closure.
async #requestLibrary(s, request) {
const { verb, args } = request;
if (Object.hasOwn(LIBRARY_LOOKUPS, verb)) return this.#requestLibraryLookup(s, request);
try {
this.#human(s);
const fields = verb === 'library.file' ? ['project', 'path'] : verb === 'library.sync' ? ['project', 'accept'] : ['project'];
@@ -5841,17 +5867,14 @@ 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,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') }));
.all('SELECT id,at,name,predecessor FROM sessions WHERE thread=? ORDER BY seq', t.id)
.map((r) => ({ ...r, default: r.id === fallback }));
}
// 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
-40
View File
@@ -1,40 +0,0 @@
-- 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;
+34 -6
View File
@@ -9,7 +9,6 @@ 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';
@@ -85,6 +84,38 @@ export async function harnessLevels({ log = (e) => process.stderr.write(`plugins
}
return (name) => levels.get(name) ?? null;
}
// Row 224 C2-05b: a host-only adapter. A failed fetch keeps serving immutable old
// objects, as C2-03a requires; no caller may select a repo, commit or sync --accept.
export function createLibraryLookup({ dataRoot, sync, now = () => Date.now() }) {
const methods = { find: libraryModule.find, get: libraryModule.get, search: libraryModule.search,
manuscript: libraryModule.manuscript, universeInfo: libraryModule.universeInfo };
return async (context, { method, args, authorize }) => {
if (!Object.hasOwn(methods, method)) throw new libraryModule.LibraryError('query-argument-invalid', 'unknown lookup');
authorize();
const source = { ...context, dataRoot };
let state = await libraryModule.libraryStatus(source);
authorize();
// sage-2819 / dewey-0852: failed attempts also pace inline retries. Otherwise an
// outage makes every lookup refetch while lastFetch remains old or null.
const recentAttempt = Math.max(state.lastFetch === null ? -Infinity : Date.parse(state.lastFetch),
state.lastError === null ? -Infinity : Date.parse(state.lastError.at));
const age = now() - recentAttempt;
if (age > libraryModule.SYNC_INTERVAL_MS || age < 0) {
try { await sync(source); }
catch (error) {
// A timer may already be fetching this mirror. Its current served state is still safe.
if (!(error instanceof libraryModule.LibraryError) || error.code !== 'library-sync-busy') throw error;
}
authorize();
state = await libraryModule.libraryStatus(source);
}
if (!state.served) throw new libraryModule.LibraryError('library-commit', 'library has never been synced');
authorize();
return methods[method]({ repo: libraryModule.mirrorPath(source), commit: state.served,
mapping: libraryModule.checkMapping(context.business.projects[context.project]) }, args);
};
}
export async function startBroker({
dataRoot,
businesses,
@@ -262,11 +293,6 @@ 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,
@@ -279,6 +305,8 @@ export async function startBroker({
store,
businesses,
library: {
lookup: createLibraryLookup({ dataRoot, sync: context =>
processing.work(() => libraryModule.sync(context, syncOptions(context.business))) }),
readFile: context => libraryModule.readLibraryFile({ ...context, dataRoot }),
status: context => libraryModule.libraryStatus({ ...context, dataRoot }),
sync: (context, { accept }) => processing.work(() => libraryModule.sync({ ...context, dataRoot }, { ...syncOptions(context.business), accept })),
+2 -9
View File
@@ -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 = '129a76db0fe9973cdea5a87f274941b4111c86730fba8a0cbe16e61a534525d0';
export const SCHEMA_VERSION = '3m';
export const SCHEMA_SHA256 = 'cd68512212ce8bf8f5631138b2e63a340fa7b940088430649bb7784af7df162a';
export const SCHEMA_VERSION = '3l';
// 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,11 +72,6 @@ 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',
@@ -94,8 +89,6 @@ 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');
+1 -1
View File
@@ -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, session: 'th-1', legacy: true },
{ thread: 'th-1', instance: 'coder', pending: 0, unread: 0, lastMessage: null, lastOpen: '2026-10-15T00:00:01.000Z', lastClaim: null, holder: null },
]);
// And its idle release (Dewey's F3, 28591): the op reaches the broker, which refuses a run
// that doesn't hold the thread.
+235
View File
@@ -0,0 +1,235 @@
// Row 224: exercise broker authorization, real C2-05a queries and host served snapshots.
// All bodies and remotes are made up. External sync is replaced only at its trusted host seam.
import test from 'node:test';
import assert from 'node:assert/strict';
import { mkdirSync, writeFileSync } from 'node:fs';
import { join } from 'node:path';
import { Store } from '../src/store.mjs';
import { Broker } from '../src/broker.mjs';
import { createLibraryLookup, startBroker } from '../src/runtime.mjs';
import { serve } from '../src/server.mjs';
import { Client } from '../src/client.mjs';
import * as library from '../../library/src/index.mjs';
import { scratchRepo, note } from '../../library/tests/query-fixture.mjs';
import { mappingEntry, git } from '../../library/tests/helpers.mjs';
const NOW = Date.parse('2026-10-22T16:00:00.000Z');
const VERBS = ['library.find', 'library.get', 'library.search', 'library.manuscript', 'library.universe'];
const GET = { universe: 'harbor', path: 'Book One/Characters/Tomas.md' };
const ARGS = [
{ universe: 'harbor', query: 'Tomas' }, GET,
{ universe: 'harbor', text: 'ferry' }, { universe: 'harbor', book: 'Book One' }, { universe: 'harbor' },
];
const TEXTS = {
'Book One/Characters/Tomas.md': note('character', 'Tomas', 'He runs the ferry. [[Elsewhere]]'),
'Book One/Chapters/01.md': '# A ferry\nOne line.\n',
'Book One/Playground/What If.md': note('character', 'Maybe Tomas', 'Unwritten ferry.'),
'Second World/Characters/Elsewhere.md': note('character', 'Elsewhere', 'Other ferry.'),
};
async function world(t, { projects, adapter, claimed = true, now = () => NOW } = {}) {
const f = scratchRepo(t, TEXTS), dataRoot = join(f.root, 'data');
mkdirSync(dataRoot);
const business = { id: 'acme', human: 'jason', vars: { 'bus.tester': true },
projects: projects ?? { books: mappingEntry() }, arbiters: { technical: 'coder', delivery: 'coder' },
roles: { coder: { definition: 'coder', authority: { withinRole: [], crossRole: [] } } } };
const context = { business, project: 'books', dataRoot };
let mirror = null, state = null, fetches = 0;
if (business.projects.books?.kind === 'library') {
mirror = await library.ensureMirror(context);
git(f.repo, '-c', 'protocol.file.allow=always', 'push', '-q', mirror, 'HEAD:refs/heads/served');
state = { version: 1, branch: 'main', served: f.commit, lastFetch: new Date(NOW).toISOString(),
lastError: null, diverged: null, problems: [] };
writeFileSync(join(mirror, library.SYNC_STATE_FILE), JSON.stringify(state), { mode: 0o600 });
}
const setState = patch => {
state = { ...state, ...patch };
writeFileSync(join(mirror, library.SYNC_STATE_FILE), JSON.stringify(state), { mode: 0o600 });
};
const lookup = createLibraryLookup({ dataRoot, now, sync: async () => {
fetches++;
setState({ lastFetch: new Date(now()).toISOString() });
return { state, code: null };
} });
const store = new Store(dataRoot);
t.after(() => store.close());
const broker = new Broker({ store, businesses: { acme: business },
library: adapter ?? { lookup, LibraryError: library.LibraryError } });
const agent = broker.bindLaunch({ business: 'acme', role: 'coder', run: 'r224', harness: 'pi' });
if (claimed) broker.request(agent, { verb: 'role.claim' });
const human = broker.bindHuman({ business: 'acme', human: 'jason', via: 'cli', outsideAgent: true });
const reader = broker.bindReader({ business: 'acme' }), tester = broker.bindTester({ business: 'acme' });
const call = (verb, args = {}, cap = agent) => Promise.resolve().then(() => broker.request(cap, { verb, args }));
return { ...f, dataRoot, business, context, mirror, store, broker, agent, human, reader, tester,
call, lookup, setState, fetches: () => fetches };
}
test('all five agent verbs read the configured served commit through real queries', async t => {
const f = await world(t);
const [find, get, search, manuscript, universe] = await Promise.all(VERBS.map((verb, i) => f.call(verb, ARGS[i])));
for (const result of [find, get, search, manuscript, universe]) assert.equal(result.commit, f.commit);
assert.deepEqual(find.results.map(x => x.path), ['Book One/Characters/Tomas.md']);
assert.match(JSON.stringify(get.lines), /He runs the ferry/);
assert.deepEqual(get.links_out, []); // The only target is in another universe.
assert.ok(search.results.length > 0);
assert.ok(search.results.every(x => !x.path.startsWith('Second World/') && x.scope === 'canon'));
assert.equal(manuscript.path, 'Book One/Chapters/01.md');
assert.deepEqual(universe.books, ['Book One']);
assert.equal(f.fetches(), 0);
const playground = await f.call('library.find', { universe: 'harbor', query: '', scope: 'playground' });
assert.deepEqual(playground.results.map(x => x.path), ['Book One/Playground/What If.md']);
assert.equal(await f.call('library.get', { universe: 'harbor', path: 'Second World/Characters/Elsewhere.md' }), null);
});
test('each verb refuses missing universe and unknown keys before host work', async t => {
let calls = 0;
const f = await world(t, { adapter: { lookup: () => { calls++; } } });
for (const [i, verb] of VERBS.entries()) {
const { universe, ...missing } = ARGS[i];
await assert.rejects(f.call(verb, missing), { code: 'invalid-request' });
for (const extra of [{ repo: '/elsewhere' }, { commit: 'b'.repeat(40) }, { business: 'other' }, { accept: 'b'.repeat(40) }])
await assert.rejects(f.call(verb, { ...ARGS[i], ...extra }), { code: 'invalid-request' });
await assert.rejects(f.call(verb, { ...ARGS[i], universe: '' }), { code: 'invalid-request' });
}
assert.equal(calls, 0);
});
test('no library, missing host adapter and undeclared project refuse library-not-configured', async t => {
const noLibrary = await world(t, { projects: {} });
const noAdapter = await world(t, { adapter: {} });
const f = await world(t);
for (const [i, verb] of VERBS.entries()) {
await assert.rejects(noLibrary.call(verb, ARGS[i]), { code: 'library-not-configured' });
await assert.rejects(noAdapter.call(verb, ARGS[i]), { code: 'library-not-configured' });
await assert.rejects(f.call(verb, { ...ARGS[i], project: 'other' }), { code: 'library-not-configured' });
}
f.business.projects.second = mappingEntry();
await assert.rejects(f.call('library.get', GET), { code: 'invalid-request' });
assert.equal((await f.call('library.get', { ...GET, project: 'books' })).commit, f.commit);
});
test('reads need the current holder on each call; human/tester/reader are not agents', async t => {
const f = await world(t, { claimed: false });
for (const [i, verb] of VERBS.entries()) {
await assert.rejects(f.call(verb, ARGS[i]), { code: 'not-holder' });
await assert.rejects(f.call(verb, ARGS[i], f.human), { code: 'agent-required' });
await assert.rejects(f.call(verb, ARGS[i], f.tester), { code: 'tester-refused' });
await assert.rejects(f.call(verb, ARGS[i], f.reader), { code: 'read-only' });
}
f.broker.request(f.agent, { verb: 'role.claim' });
assert.equal((await f.call('library.get', GET)).commit, f.commit);
f.broker.request(f.agent, { verb: 'role.release' });
await assert.rejects(f.call('library.get', GET), { code: 'not-holder' });
});
test('releasing a claim during an async lookup fences the response', { timeout: 30_000 }, async t => {
let release, entered;
const started = new Promise(resolve => { entered = resolve; });
const f = await world(t, { adapter: { lookup: async () => {
entered(); await new Promise(resolve => { release = resolve; }); return { lines: [[1, 'not returned']] };
} } });
const pending = f.call('library.get', GET);
await started;
f.broker.request(f.agent, { verb: 'role.release' });
release();
await assert.rejects(pending, { code: 'not-holder' });
});
test('a business library removed during an async read cannot leak its result', { timeout: 30_000 }, async t => {
let release, entered;
const started = new Promise(resolve => { entered = resolve; });
const f = await world(t, { adapter: { lookup: async () => {
entered(); await new Promise(resolve => { release = resolve; }); return { lines: [[1, 'not returned']] };
} } });
const pending = f.call('library.get', GET);
await started;
delete f.business.projects.books;
release();
await assert.rejects(pending, { code: 'library-not-configured' });
});
test('freshness boundary, unknown last fetch and failed fetch still use immutable served state', async t => {
const f = await world(t);
f.setState({ lastFetch: new Date(NOW - 60_000).toISOString() });
await f.call('library.get', GET);
assert.equal(f.fetches(), 0);
f.setState({ lastFetch: new Date(NOW - 60_001).toISOString() });
await f.call('library.get', GET);
assert.equal(f.fetches(), 1);
f.setState({ lastFetch: null });
await f.call('library.universe', { universe: 'harbor' });
assert.equal(f.fetches(), 2);
let attempted = 0, clock = NOW;
const failing = createLibraryLookup({ dataRoot: f.dataRoot, now: () => clock, sync: async () => {
attempted++;
f.setState({ lastError: { code: 'library-fetch-failed', at: new Date(clock).toISOString() } });
return { code: 'library-fetch-failed' };
} });
f.setState({ lastFetch: null });
const result = await failing(f.context, { method: 'get', args: GET, authorize() {} });
assert.equal(attempted, 1);
assert.equal(result.commit, f.commit);
for (const elapsed of [1, 59_999, 60_000]) {
clock = NOW + elapsed;
const retry = await failing(f.context, { method: 'get', args: GET, authorize() {} });
assert.equal(retry.commit, f.commit, 'serve the old pin during the failure cooldown');
assert.equal(attempted, 1, 'recent lastError prevents another sync even though lastFetch is null');
}
clock = NOW + 60_001;
assert.equal((await failing(f.context, { method: 'get', args: GET, authorize() {} })).commit, f.commit);
assert.equal(attempted, 2, 'retry only after the failed-attempt interval expires');
});
test('an unreadable sync state fails closed; a busy sync keeps serving the existing pin', async t => {
const f = await world(t);
const lookup = createLibraryLookup({ dataRoot: f.dataRoot, now: () => NOW, sync: async () => {
throw new library.LibraryError('library-sync-busy', 'another fetch');
} });
f.setState({ lastFetch: null });
assert.equal((await lookup(f.context, { method: 'get', args: GET, authorize() {} })).commit, f.commit);
writeFileSync(join(f.mirror, library.SYNC_STATE_FILE), '{bad json', { mode: 0o600 });
await assert.rejects(f.call('library.get', GET), { code: 'library-state-invalid' });
});
test('a stale lookup sees the commit published by its sync, not the prior served commit', async t => {
const f = await world(t);
writeFileSync(join(f.repo, GET.path), note('character', 'Tomas', 'Changed ferry.'));
git(f.repo, 'add', '-A'); git(f.repo, 'commit', '-qm', 'second fixture');
const next = git(f.repo, 'rev-parse', 'HEAD');
const lookup = createLibraryLookup({ dataRoot: f.dataRoot, now: () => NOW, sync: async () => {
git(f.repo, '-c', 'protocol.file.allow=always', 'push', '-q', f.mirror, 'HEAD:refs/heads/served');
f.setState({ served: next, lastFetch: new Date(NOW).toISOString() });
return { code: null };
} });
f.setState({ lastFetch: null });
const result = await lookup(f.context, { method: 'get', args: GET, authorize() {} });
assert.equal(result.commit, next);
assert.match(JSON.stringify(result.lines), /Changed ferry/);
});
test('real socket awaits lookup results and rechecks rejected query arguments', { timeout: 30_000 }, async t => {
const f = await world(t);
const path = join(f.root, 'lookup.sock');
const server = await serve({ broker: f.broker, path }); t.after(() => server.close());
const client = new Client({ path, cap: f.agent });
const result = await client.call('library.get', GET);
assert.equal(result.commit, f.commit);
await assert.rejects(client.call('library.find', { universe: 'harbor', query: '', scope: 'anything' }), /query-argument-invalid/);
});
test('runtime actually injects the lookup adapter, without exposing human library.file', { timeout: 30_000 }, async t => {
const f = scratchRepo(t, TEXTS), dataRoot = join(f.root, 'runtime'); mkdirSync(dataRoot);
const business = { id: 'acme', human: 'jason', projects: { books: mappingEntry() },
arbiters: { technical: 'coder', delivery: 'coder' },
roles: { coder: { definition: 'coder', authority: { withinRole: [], crossRole: [] } } } };
const runtime = await startBroker({ dataRoot, businesses: { acme: business } }); t.after(() => runtime.close());
const mirror = library.mirrorPath({ business, project: 'books', dataRoot });
git(f.repo, '-c', 'protocol.file.allow=always', 'push', '-q', mirror, 'HEAD:refs/heads/served');
writeFileSync(join(mirror, library.SYNC_STATE_FILE), JSON.stringify({ version: 1, branch: 'main', served: f.commit,
lastFetch: new Date().toISOString(), lastError: null, diverged: null, problems: [] }), { mode: 0o600 });
const cap = runtime.broker.bindLaunch({ business: 'acme', role: 'coder', run: 'r224-runtime', harness: 'pi' });
runtime.broker.request(cap, { verb: 'role.claim' });
assert.equal((await runtime.broker.request(cap, { verb: 'library.get', args: GET })).commit, f.commit);
await assert.rejects(runtime.broker.request(cap, { verb: 'library.file', args: { project: 'books', path: GET.path } }),
{ code: 'human-required' });
});
@@ -1,98 +0,0 @@
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' });
});
+7 -7
View File
@@ -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, run) => {
const refusedOnce = (fn, label) => {
const from = lastSeq();
assert.throws(fn, { code: 'not-holder' }, label);
const added = store.all('SELECT kind, actor_run, body FROM events WHERE seq > ? ORDER BY seq', from);
const added = store.all('SELECT kind, body FROM events WHERE seq > ? ORDER BY seq', from);
assert.deepEqual(
added.map((e) => [e.kind, e.actor_run, JSON.parse(e.body)]),
[['action.refused', run, { code: 'not-holder' }]],
added.map((e) => [e.kind, JSON.parse(e.body)]),
[['action.refused', { code: 'not-holder' }]],
label,
);
};
const refused = (name, { d, taken }) => {
const { events, ...before } = counts();
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`);
refusedOnce(() => call(name, 'decision.withdraw', { id: d.id }), 'approval');
refusedOnce(() => call(name, 'message.receive'), 'delivery');
refusedOnce(() => call(name, 'message.read', { id: taken.id }), 'read');
const { events: after, ...rest } = counts();
assert.deepEqual(rest, before);
assert.equal(after, events + 3);
+51 -136
View File
@@ -75,15 +75,6 @@
// 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
@@ -130,14 +121,12 @@
// 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>/, 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.
// workspaces/<b>/.retired/I-<transaction>/. 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, relative } from "node:path";
import { dirname, join } 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";
@@ -153,16 +142,13 @@ 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";
@@ -251,9 +237,6 @@ 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;
@@ -284,22 +267,6 @@ 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(() => {});
@@ -311,7 +278,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, session = null, human = false } = {}) {
async function check(instance, { pm = false, thread = 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`);
@@ -319,12 +286,7 @@ 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 (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`);
if (registry.running(b.id, instance, thread)) throw new Refusal("instance-running");
let resolved;
try {
resolved = resolveInstance({ system: vars, business: b, instance });
@@ -416,59 +378,29 @@ export function createLauncher({
}
// Bundle, spawn, bind, register, hand over the capability.
async function start({ instance, launchedBy, resolved, model, account = null, thread = null, session = null, legacy = false, legacyThread = legacy, human = false, admissionGeneration }) {
async function start({ instance, launchedBy, resolved, model, account = null, thread = null, 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 session's store is checked first, so a bad id or an unsafe
// directory launches nothing.
// A thread's store is checked first, so a bad id or an unsafe directory
// launches nothing.
let store = null;
if (thread) {
try {
store = sessionStore(dataRoot, b.id, thread, session, { legacy, legacyThread });
store = threadStore(dataRoot, b.id, thread);
} catch (e) {
throw new Refusal(e.code === "legacy-store-unmapped" ? e.code : "thread-store-refused", e.message);
throw new Refusal("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 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;
// 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 });
let checkouts;
try {
checkouts = await withProcessing(dataRoot, () => {
@@ -484,12 +416,11 @@ 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,session,human}));
({resolved,model,account,harness,route,routeReason}=await check(instance,{pm:instance===b.launch.by,thread,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 }));
@@ -508,8 +439,9 @@ export function createLauncher({
contract: readFileSync(resolved.contract, "utf8"),
arbiter: Object.values(b.arbiters).includes(instance),
tracker,
library: Object.values(b.projects ?? {}).some(project => project.kind === "library"),
credentials,
session: { run, launchedBy, workspace: inSandbox, toolSocket, turnMarker, checkouts: inSandbox === workspace ? checkouts : checkouts.map((c) => ({ ...c, path: join(inSandbox, relative(workspace, c.path)) })) },
session: { run, launchedBy, workspace, toolSocket, turnMarker, checkouts },
harness,
route,
routeReason,
@@ -540,7 +472,7 @@ export function createLauncher({
brokerSocket: host.path,
launchSocket: launchSocketPath(dataRoot),
gitSocket: git,
workspace: inSandbox,
workspace,
sessionDir,
adapter,
bundle: files,
@@ -553,7 +485,6 @@ export function createLauncher({
const paths = {
workspace,
workspaceDest,
adapter,
harnessMounts,
brokerSocket: host.path,
@@ -589,12 +520,11 @@ 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");
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 } : {}) });
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 } : {}) });
bound = true;
admission(admissionGeneration);
entry = {
business: b.id, instance, ...(thread ? { thread, session, workspace } : {}), run, launchedBy, harness: model.harness, model: model.model, provider: model.provider, modelKey: model.modelKey,
business: b.id, instance, ...(thread ? { thread } : {}), 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);
@@ -623,16 +553,13 @@ 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, session } : {}), run, reason, exitCode: code, signal, ...(haltGeneration ? { haltGeneration } : {}), ...(ended?.limit ? { limit: ended.limit } : {}) });
record({ event: "end", instance, ...(thread ? { thread } : {}), 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);
@@ -640,7 +567,7 @@ export function createLauncher({
refusedRelease.delete(run);
}
});
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 });
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 });
return { run, instance, harness: model.harness, model: model.model };
}
@@ -674,12 +601,9 @@ 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 key = `${t.thread}/${t.session}`;
const s = registry.running(b.id, t.instance, t.thread, t.session);
const s = registry.running(b.id, t.instance, t.thread);
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) {
@@ -693,7 +617,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, session: t.session, run: s.run });
record({ event: "thread-idle", instance: t.instance, thread: t.thread, run: s.run });
try {
if (startTimeOf(s.runnerPid) === s.runnerStartTime) process.kill(s.runnerPid, "SIGTERM");
} catch {}
@@ -706,21 +630,20 @@ 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(key) === signature) continue;
if (startedFor.get(t.thread) === signature) continue;
await serialize(async () => {
if (closed || registry.running(b.id, t.instance, t.thread, t.session)) return;
if (closed || registry.running(b.id, t.instance, t.thread)) return;
try {
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);
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);
} catch (e) {
const code = e instanceof Refusal ? e.code : "launch-failed";
// Logged once per code, not once per tick.
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}:{}) });
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}:{}) });
}
});
}
@@ -927,23 +850,20 @@ export function createLauncher({
const transaction = out?.transaction ?? null;
if (transaction === null) return { refusal: "workspace-rename-failed" };
const bot = out.bot ? { bot: out.bot } : {};
// 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 });
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}` };
}
record({ event: "workspace-retired", instance: r.instance, request: r.request, transaction, path: to });
return { transaction, ...bot };
}
@@ -1039,15 +959,10 @@ export function createLauncher({
process.kill(s.pid, "SIGKILL");
} catch {}
const end = Date.now() + 5000;
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`);
while (startTimeOf(s.pid) === s.startTime && Date.now() < end) await new Promise((r) => setTimeout(r, 20));
}
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 } : {}), ...(s.session ? { session: s.session } : {}) });
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 } : {}) });
} 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`);
@@ -1055,7 +970,7 @@ export function createLauncher({
continue;
}
await endRun(s.run, "host-lost", 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 });
record({ event: "end", instance: s.instance, ...(s.thread ? { thread: s.thread } : {}), run: s.run, reason: "host-lost", exitCode: null, signal: null });
log(`run ${s.run} (${s.instance}) from an earlier host ended: host-lost`);
}
leftover = [];
-18
View File
@@ -1,18 +0,0 @@
// 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");
-414
View File
@@ -1,414 +0,0 @@
// 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");
});
+23 -2
View File
@@ -27,6 +27,7 @@ import { bootConfig, loadSystem } from "../src/config.mjs";
import { startTimeOf } from "../src/host.mjs";
import { createLauncher } from "../src/launcher.mjs";
import { fakeHarnesses, fixture, tmp } from "./helpers.mjs";
import { mappingEntry } from "../../library/tests/helpers.mjs";
// Row 264a (#1769): a thread made by hand gets the default session the 3k-3l migration gives a
// legacy thread, with the thread's id.
const DEFAULT_SESSION = [
@@ -179,11 +180,10 @@ 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, "sessions", host.runtime.threadsState("acme").find((x) => x.thread === A).session);
const store = join(f.dataRoot, "threads", "acme", A);
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.
@@ -443,3 +443,24 @@ test("REQ-LIMIT-4 a thread run at its token limit stops at the end of its turn:
assert.deepEqual(threadRuns(f, A), []);
assert.equal(host.runtime.threadsState("acme").find((x) => x.thread === A).holder, null);
});
// Row224: the launcher must derive library presence from business projects, not
// from session checkouts (a library never has an agent clone).
test("C2-05b a real launched bundle gets lookups for a library project without a library checkout", { skip, timeout: 60000 }, async t => {
const prepared = prepare(t, { idle: 10, edit: doc => {
doc.projects.books = mappingEntry();
return doc;
} });
const { f } = prepared;
const host = await start(t, prepared);
const project = host.call("project.create", { name: "library work", repo: "stack" }).id;
const thread = host.call("thread.create", { project, name: "lookup" }).id;
await ask(host.call, thread, "inspect bundle");
const run = launchLog(f).find(e => e.event === "launch" && e.thread === thread).run;
const tools = JSON.parse(readFileSync(join(f.dataRoot, "launches", "acme", run, "bundle", "tools.json"), "utf8"));
assert.deepEqual(tools.filter(tool => tool.name.startsWith("library_")).map(tool => tool.name),
["library_find", "library_get", "library_search", "library_manuscript", "library_universe"]);
const [session] = threadRuns(f, thread);
assert.ok(session);
assert.ok(!session.checkouts?.some(checkout => checkout.project === "books"));
});
+3 -2
View File
@@ -292,6 +292,7 @@ function prompt({ contract, resolved, session, tools, untyped, builtins, persona
// resolved resolveInstance(...) output
// contract the role contract's text
// arbiter true when the business names this instance as an arbiter
// library true when the trusted business configuration has a library project
// tracker true when the broker has task verbs for the business
// credentials the broker's credentialStatus(...) for the instance (metadata only)
// session { run, launchedBy, workspace, toolSocket, turnMarker, checkouts }
@@ -309,11 +310,11 @@ function prompt({ contract, resolved, session, tools, untyped, builtins, persona
// (row 213); an instance with persona.sha256 and no personaDir refuses
// Returns { manifest, files, policy, tools, env } with absolute paths; env is the module's
// adapter variables, which the host writes into session.json.
export function buildBundle({ dir, resolved, contract, arbiter = false, tracker = false, credentials = [], session, harness, route = "primary", routeReason = null, harnessVersion = null, models = null, land = false, skillsDir = null, personaDir = null }) {
export function buildBundle({ dir, resolved, contract, arbiter = false, tracker = false, library = false, credentials = [], session, harness, route = "primary", routeReason = null, harnessVersion = null, models = null, land = false, skillsDir = null, personaDir = null }) {
if (route !== "primary" && route !== "fallback") throw new BundleError(`route ${JSON.stringify(route)} isn't primary or fallback`);
const model = sessionModel(resolved, models, route);
const web = webPlan(resolved, model.harness, harness.builtinWeb());
const tools = typedTools({ resolved, arbiter, tracker, projects: (session.checkouts ?? []).map((c) => c.project), webTyped: !web.builtin, land });
const tools = typedTools({ resolved, arbiter, tracker, library, projects: (session.checkouts ?? []).map((c) => c.project), webTyped: !web.builtin, land });
const untyped = untypedActions(resolved, tools);
const persona = readPersona(resolved, personaDir);
mkdirSync(dir, { recursive: true, mode: 0o700 });
+34 -2
View File
@@ -78,9 +78,10 @@ const TASK_TOOLS = {
const MESSAGE_CLASSES = ["REQUEST", "ASSIGNMENT", "REVIEW-REQUEST", "REVIEW-RESULT", "RESULT", "INFO", "DECISION", "REACTION", "BLOCK"];
// typedTools({ resolved, arbiter, tracker, projects, land }) -> [{ name, verb, action, description, parameters }]
// typedTools({ resolved, arbiter, tracker, projects, land, library }) -> [{ name, verb, action, description, parameters }]
// resolved resolveInstance(...) output for the session's instance
// arbiter true when the business names this instance as an arbiter
// library true when the trusted business configuration has a library project
// tracker true when the broker has task verbs for the business
// projects the project ids the session has a clone of
// land true when the role and thread hold a merge decision a landing could use (landHeld)
@@ -88,7 +89,7 @@ const MESSAGE_CLASSES = ["REQUEST", "ASSIGNMENT", "REVIEW-REQUEST", "REVIEW-RESU
// `verb` is the broker verb; the `launch` tool has verb "launch" and goes
// to the host instead, and "working.push", "working.fetch" and
// "working.land" go to the host's git socket.
export function typedTools({ resolved, arbiter = false, tracker = false, projects = [], webTyped = true, land = false }) {
export function typedTools({ resolved, arbiter = false, tracker = false, projects = [], webTyped = true, land = false, library = false }) {
const held = new Set([...resolved.limits.authority.withinRole, ...resolved.limits.authority.crossRole]);
const tools = [];
const add = (name, verb, action, description, parameters) => tools.push({ name, verb, action, description, parameters });
@@ -222,6 +223,32 @@ export function typedTools({ resolved, arbiter = false, tracker = false, project
url: str("The https url to fetch", { maxLength: 2048 }),
}, ["url"]));
}
// Row224: no library clone or token reaches a session. The broker chooses the served
// commit and requires a universe on every call; cursors page without cutting replies.
if (library) {
const universe = str("Universe ID in the configured library", { minLength: 1 });
const project = str("Library project ID; omit only when the business has one library");
const scope = str("canon by default; playground is opt-in", { enum: ["canon", "playground"] });
const book = str("Book folder from library_universe");
const offset = { type: "integer", minimum: 0, description: "next from the previous page" };
const line = { type: "integer", minimum: 1, description: "One-based line number" };
const base = { universe, project };
add("library_find", "library.find", null, "Find names in one universe, at the served commit. At most 50 results; continue with next as offset.", obj({
...base, query: str("Case-insensitive name or filename substring; empty matches all"),
type: str("Mapped type", { enum: ["character", "place", "research", "tracker", "summary"] }),
book, scope, offset,
}, ["universe", "query"]));
add("library_get", "library.get", null, "Read a mapped note and visible links in one universe at the served commit. next pages lines; links_next pages links. An explicit own-universe playground path is readable.", obj({
...base, path: str("Mapped note path relative to the library"), from: line, links_from: offset,
}, ["universe", "path"]));
add("library_search", "library.search", null, "Search text in one universe at the served commit. At most 50 hits; use next as offset. Playground is opt-in.", obj({
...base, text: str("Nonempty case-insensitive substring", { minLength: 1 }), book, scope, offset,
}, ["universe", "text"]));
add("library_manuscript", "library.manuscript", null, "Read one chapter of a book at the served commit, at most 400 lines. Omit chapter for the first; continue with next or next_chapter.", obj({
...base, book, scope, chapter: str("Chapter filename without extension"), from: line, to: line,
}, ["universe", "book"]));
add("library_universe", "library.universe", null, "Read a universe's mapping at the served commit: books, types, playground prefixes, and hybrid sources/constraints/notes or draft metadata.", obj(base, ["universe"]));
}
add("read_spec", "spec.read", null, "Read a project's committed north-star spec and its file history. commit is an optional full reachable commit ID. The file commit, rather than HEAD, is what a decision cites. confirmed says the file commit was authored by the business's human.", obj({
project: str("Project ID from the business file"),
commit: str("Optional full historical commit ID", { pattern: "^[0-9a-f]{40}$" }),
@@ -277,6 +304,11 @@ export function typedTools({ resolved, arbiter = false, tracker = false, project
if (held.has("task.create"))
add("list_requirements", "requirements", null, "List the approved PRD's requirement ids, each with a one-line title. task_create takes only these. When none fits a request, raise a prd.approve decision instead of creating the task.", obj({}));
}
// mos-1308 / M3: no generic spec, evidence, tracker, decision or write tools. The
// C2-08 web grant still supplies broker web reads to companion and research only.
if (CREATIVE_DEFINITIONS.includes(resolved.definition))
return tools.filter(tool => tool.name === "send_message" || tool.name.startsWith("library_") ||
tool.name === "web_search" || tool.name === "web_fetch");
return tools;
}
@@ -0,0 +1,71 @@
// Row224 M3: exact creative tool sets preserve the existing C2-08 web grants.
import test from 'node:test';
import assert from 'node:assert/strict';
import { mkdirSync, readFileSync } from 'node:fs';
import { join } from 'node:path';
import { typedTools } from '../src/tools.mjs';
import { buildBundle } from '../src/bundle.mjs';
import { TYPES } from '../../library/src/mapping.mjs';
import { HARNESS, REPO_SKILLS, resolvedFor, scratch } from './helpers.mjs';
const LOOKUPS = ['library_find', 'library_get', 'library_search', 'library_manuscript', 'library_universe'];
const CREATIVE = ['companion', 'research', 'ideas', 'proofreader', 'fact-checker', 'continuity'];
const role = (t, definition, harness = 'pi') => resolvedFor(t, 'bean', d => {
d.roles.bean = { definition, vars: { harness, model: 'claude-sonnet-5-5', thinking: 'high' } };
});
const expected = definition => ['send_message', ...(['companion', 'research'].includes(definition)
? ['web_search', 'web_fetch'] : []), ...LOOKUPS];
test('M3: per-role snapshots contain six names, or eight with the C2-08 web grant', t => {
for (const definition of CREATIVE) {
const tools = typedTools({ resolved: role(t, definition), library: true, tracker: true, arbiter: true });
assert.deepEqual(tools.map(t => t.name), expected(definition), definition);
assert.ok(tools.every(t => t.parameters.additionalProperties === false));
}
});
test('every noncreative role gets lookups when a library is configured, not otherwise', t => {
for (const instance of ['coder', 'reviewer', 'pm', 'chief']) {
const resolved = resolvedFor(t, instance);
const ordinary = typedTools({ resolved }).map(t => t.name);
assert.ok(!ordinary.some(name => name.startsWith('library_')));
const tools = typedTools({ resolved, library: true });
assert.deepEqual(tools.filter(t => t.name.startsWith('library_')).map(t => t.name), LOOKUPS);
assert.deepEqual(tools.filter(t => !t.name.startsWith('library_')).map(t => t.name), ordinary);
}
});
test('lookup schemas require a universe even for get, include paging, and cannot select repo or commit', t => {
const tools = typedTools({ resolved: role(t, 'ideas'), library: true });
const required = [ ['universe', 'query'], ['universe', 'path'], ['universe', 'text'], ['universe', 'book'], ['universe'] ];
for (const [i, tool] of tools.filter(t => t.name.startsWith('library_')).entries()) {
assert.deepEqual(tool.parameters.required, required[i]);
assert.equal(tool.verb, LOOKUPS[i].replace('_', '.'));
assert.equal(tool.action, null);
assert.equal(tool.parameters.additionalProperties, false);
for (const key of ['repo', 'commit', 'business', 'accept']) assert.equal(Object.hasOwn(tool.parameters.properties, key), false);
}
const byName = Object.fromEntries(tools.map(t => [t.name, t]));
assert.deepEqual(byName.library_find.parameters.properties.scope.enum, ['canon', 'playground']);
assert.deepEqual(byName.library_find.parameters.properties.type.enum, [...TYPES]);
assert.equal(byName.library_find.parameters.properties.offset.minimum, 0);
assert.equal(byName.library_get.parameters.properties.from.minimum, 1);
assert.equal(byName.library_get.parameters.properties.links_from.minimum, 0);
assert.equal(byName.library_manuscript.parameters.properties.chapter.type, 'string');
});
test('bundle flag reaches both harness tool manifests and creative prompt without generic tools', t => {
for (const definition of CREATIVE) for (const harness of ['pi', 'claude-code']) {
const dir = scratch(t), workspace = join(dir, 'ws'); mkdirSync(workspace);
const b = buildBundle({ dir: join(dir, 'bundle'), resolved: role(t, definition, harness), library: true,
contract: '# Creative\n', tracker: true, arbiter: true, skillsDir: REPO_SKILLS,
session: { run: 'r224', launchedBy: 'chief', workspace, toolSocket: join(dir, 't.sock'), turnMarker: join(dir, 'turn.done') },
harness: HARNESS[harness] });
assert.deepEqual(b.tools.map(t => t.name), expected(definition), `${definition} ${harness}`);
const serialized = JSON.parse(readFileSync(b.files.tools, 'utf8'));
assert.deepEqual(serialized.map(t => t.name), expected(definition));
const prompt = readFileSync(b.files.prompt, 'utf8');
assert.match(prompt, /library_find/);
assert.doesNotMatch(prompt, /Bus tools: [^\n]*(read_spec|list_agents|inbox|trail|write_record|raise_decision|list_tasks)/);
}
});
+75
View File
@@ -0,0 +1,75 @@
// The lookup index of design 3 (C2-05a): types, names, links and backlinks for one commit of her repo.
// It is built from the reader's output, so it holds only mapped, readable files: R1, R2 and R7 leave a
// file out, and an unresolved link is not in it. It is cached in memory by commit id. The key also
// carries the repo path and a digest of the mapping, because one commit reads differently under another
// mapping. The index is not universe-aware; query.mjs decides what a universe may see.
import { createHash } from 'node:crypto';
import { readLibrary } from './reader.mjs';
import { raise } from './problems.mjs';
const CACHE_LIMIT = 8;
const cache = new Map();
const push = (map, key, value) => {
if (!map.has(key)) map.set(key, []);
map.get(key).push(value);
};
// A file's name for lookups: its front matter `name` when it is text, else its file name.
const displayName = file => {
const name = file.frontmatter?.name;
return typeof name === 'string' && name.trim() ? name.trim() : file.name;
};
// A link as the index keeps it: the file at the other end, its line, and the field when front matter holds it.
const linkAt = (path, link) => ({ path, line: link.line, ...(link.field === undefined ? {} : { field: link.field }) });
// Builds the index from readLibrary's output. Pure: no git, no clock.
export function buildIndex(library, mapping) {
const files = new Map(), types = new Map(), names = new Map(), linksOut = new Map(), linksIn = new Map();
for (const file of library.files) {
const name = displayName(file);
files.set(file.path, {
path: file.path, scope: file.scope, universe: file.universe, book: file.book, type: file.type,
folder: file.folder, stem: file.name, name, frontmatter: file.frontmatter, lines: file.lines,
});
if (file.type !== null) push(types, file.type, file.path);
push(names, name.toLowerCase(), file.path);
linksOut.set(file.path, file.links.map(link => linkAt(link.path, link)));
for (const link of file.links) push(linksIn, link.path, linkAt(file.path, link));
}
return Object.freeze({
commit: library.commit, mapping, files, paths: [...files.keys()], types, names, linksOut, linksIn,
problems: library.problems,
});
}
const digest = mapping => createHash('sha256').update(JSON.stringify(mapping)).digest('hex');
// The index for one commit under one mapping, read once and kept in memory. A repeat call returns the same
// object. Past CACHE_LIMIT entries the oldest goes. A failed read is not kept.
export async function indexFor({ repo, commit, mapping }) {
if (!mapping || mapping.kind !== 'library') raise('library-mapping-invalid', 'expected a checked library mapping');
const key = JSON.stringify([repo, commit, digest(mapping)]);
if (cache.has(key)) {
const hit = cache.get(key);
cache.delete(key);
cache.set(key, hit);
return hit;
}
const pending = readLibrary({ repo, commit, mapping }).then(library => buildIndex(library, mapping));
cache.set(key, pending);
while (cache.size > CACHE_LIMIT) cache.delete(cache.keys().next().value);
try {
return await pending;
} catch (error) {
if (cache.get(key) === pending) cache.delete(key);
throw error;
}
}
export function clearIndexCache() {
cache.clear();
}
export function indexCacheSize() {
return cache.size;
}
+2
View File
@@ -8,3 +8,5 @@ export { renderItems, KINDS, NOTICES, UNREADABLE } from './render.mjs';
export { survey, formatSurvey, surveyMain } from './survey.mjs';
export { SYNC_STATE_FILE, mirrorPath, ensureMirror, withLibraryLock, servedCommit, readSyncState, libraryStatus, readLibraryFile } from './mirror.mjs';
export { syncCommand, sync, ASKPASS, SLOT, SYNC_INTERVAL_MS } from './sync.mjs';
export { buildIndex, indexFor, clearIndexCache, indexCacheSize } from './index-build.mjs';
export { sees, find, get, search, manuscript, universeInfo, QUERY_REPLY_CAP, LOOKUP_LIMIT, MANUSCRIPT_LINES } from './query.mjs';
+240
View File
@@ -0,0 +1,240 @@
// The lookup queries of design 3 (C2-05a): find, get, search, manuscript and universeInfo. Each takes a
// source {repo, commit, mapping} and an argument object, and returns a reply that fits QUERY_REPLY_CAP.
// Universe isolation is here, in sees(): a file reaches a reply only when sees() allows it. Scope comes
// from the path (D2). Unmapped files are not in the index, so they never appear. Lists page by position
// and never cut an item. A search line too long to fit alone comes back as a marked hit with no excerpt;
// a find item, a file's front matter or one of its lines that cannot fit alone is refused.
import { raise } from './problems.mjs';
import { TYPES } from './mapping.mjs';
import { indexFor } from './index-build.mjs';
// The same cap as packages/harness/src/tools.mjs REPLY_CAP. The query tests assert the two stay equal.
export const QUERY_REPLY_CAP = 40_000;
export const LOOKUP_LIMIT = 50;
export const MANUSCRIPT_LINES = 400;
const SCOPES = Object.freeze(['canon', 'playground']);
const bytes = value => JSON.stringify(value).length;
const invalid = message => raise('query-argument-invalid', message);
const inside = (folder, path) => path.startsWith(folder.endsWith('/') ? folder : `${folder}/`);
function textArg(value, name) {
if (typeof value !== 'string') invalid(`${name} must be text`);
return value;
}
function scopeArg(value) {
if (value === undefined) return 'canon';
if (!SCOPES.includes(value)) invalid('scope must be "canon" or "playground"');
return value;
}
function positionArg(value, name, fallback) {
if (value === undefined) return fallback;
if (!Number.isSafeInteger(value) || value < 1) invalid(`${name} must be a line number of 1 or more`);
return value;
}
function offsetArg(value) {
if (value === undefined) return 0;
if (!Number.isSafeInteger(value) || value < 0) invalid('offset must be a whole number of 0 or more');
return value;
}
function universeArg(mapping, value) {
if (typeof value !== 'string' || !value) invalid('universe is required');
if (!Object.hasOwn(mapping.universes, value)) raise('query-universe-unknown', `this library has no universe ${JSON.stringify(value)}`);
return value;
}
// Whether universe U may see a file. Its own files, in either scope. A hybrid also sees the canon of
// its declared sources, minus the folders its exclude constraints name. A hybrid never sees a source's
// playground. Pin constraints do not hide anything; they are reported by universeInfo only.
export function sees(mapping, universe, file) {
if (file.universe === universe) return true;
const entry = mapping.universes[universe];
if (entry.kind !== 'hybrid' || !entry.sources.includes(file.universe) || file.scope !== 'canon') return false;
return !entry.constraints.some(c => c.exclude !== undefined && inside(c.exclude, file.path));
}
// The files one universe may see in one scope, with optional book and type filters, in path order.
async function visible(source, universe, { scope, book, type }) {
const index = await indexFor(source);
const files = index.paths
.map(path => index.files.get(path))
.filter(file => file.scope === scope && sees(index.mapping, universe, file)
&& (book === undefined || file.book === book) && (type === undefined || file.type === type));
return { index, files };
}
// A reply whose size is over the cap is refused, never cut.
function checked(reply) {
if (bytes(reply) > QUERY_REPLY_CAP) raise('query-item-too-large', `the reply is over the ${QUERY_REPLY_CAP}-character cap`);
return reply;
}
// Picks items from `offset`, at most `limit`, while the reply still fits. build(picked, next) returns
// the reply; `next` is the position of the first item not picked, or null when none is left. An item
// that cannot fit even alone is refused: skipping it would hide it, and cutting it would break it.
function fitPage(items, { offset = 0, limit = LOOKUP_LIMIT, build }) {
const picked = [];
let at = offset;
while (at < items.length && picked.length < limit) {
// The size check uses items.length for `next`, the widest number a cursor can take, so a page never runs over.
if (bytes(build([...picked, items[at]], items.length)) > QUERY_REPLY_CAP) {
if (!picked.length) raise('query-item-too-large', `the item at position ${at} alone is over the ${QUERY_REPLY_CAP}-character cap`);
break;
}
picked.push(items[at]);
at += 1;
}
return checked(build(picked, at < items.length ? at : null));
}
// find: the files whose name or file name holds `query`, case-insensitive. Up to 50 per page.
export async function find(source, { universe: name, query, type, book, scope, offset } = {}) {
const universe = universeArg(source.mapping, name);
const needle = textArg(query, 'query').toLowerCase();
if (type !== undefined && !TYPES.includes(type)) invalid(`type must be one of ${TYPES.join(', ')}`);
const { index, files } = await visible(source, universe, {
scope: scopeArg(scope), book: book === undefined ? undefined : textArg(book, 'book'), type,
});
const hits = files.filter(file => file.name.toLowerCase().includes(needle) || file.stem.toLowerCase().includes(needle));
return fitPage(hits, {
offset: offsetArg(offset),
build: (picked, next) => ({
commit: index.commit,
results: picked.map(file => ({
path: file.path, type: file.type, name: file.name, appears_in: file.frontmatter?.appears_in ?? null,
scope: file.scope, universe: file.universe, book: file.book,
})),
next,
}),
});
}
// get: one visible file, its front matter, its lines from `from`, and its visible links both ways.
// Returns null when the file is missing or not visible, so a caller cannot tell the two apart. Lines page
// by `from` (reply `next`), links page together by `links_from` (reply `links_next`); `links_out` come
// before `links_in`. Every page carries the front matter. Links take at most half the cap, so a hub note
// with many backlinks stays readable and still leaves room for lines. A page with lines left carries at
// least one line, so links shrink when a line needs the room; read on with `from: next` and
// `links_from: links_next`. A `from` past the last line returns links alone. Only front matter, or a
// line that cannot fit with no links beside it, is refused.
export async function get(source, { universe: name, path, from, links_from: linksFrom } = {}) {
const universe = universeArg(source.mapping, name);
const startLine = positionArg(from, 'from', 1);
const linkStart = offsetArg(linksFrom);
if (typeof path !== 'string') invalid('path must be text');
const index = await indexFor(source);
const file = index.files.get(path);
if (!file || !sees(index.mapping, universe, file)) return null;
const reachable = links => links.filter(link => {
const other = index.files.get(link.path);
return other !== undefined && sees(index.mapping, universe, other);
});
const outs = reachable(index.linksOut.get(file.path) ?? []);
const ins = reachable(index.linksIn.get(file.path) ?? []);
const links = [...outs.map(link => ({ out: true, link })), ...ins.map(link => ({ out: false, link }))];
const lines = file.lines.slice(startLine - 1);
const linkRest = Math.max(0, links.length - linkStart);
const page = (nLines, nLinks) => {
const taken = links.slice(linkStart, linkStart + nLinks);
return {
path: file.path, commit: index.commit, scope: file.scope, frontmatter: file.frontmatter,
lines: lines.slice(0, nLines),
links_out: taken.filter(item => item.out).map(item => item.link),
links_in: taken.filter(item => !item.out).map(item => item.link),
next: nLines < lines.length ? startLine + nLines : null,
links_next: nLinks < linkRest ? linkStart + nLinks : null,
};
};
// How many leading items fit in `room`, by each item's own size.
const count = (items, room) => {
let used = 0, n = 0;
while (n < items.length && used + bytes(items[n]) + 1 <= room) used += bytes(items[n++]) + 1;
return n;
};
const base = bytes(page(0, 0));
if (base > QUERY_REPLY_CAP) raise('query-item-too-large', `the front matter of ${file.path} alone is over the ${QUERY_REPLY_CAP}-character cap`);
if (lines.length && bytes(page(1, 0)) > QUERY_REPLY_CAP) {
raise('query-item-too-large', `line ${lines[0][0]} of ${file.path} alone is over the ${QUERY_REPLY_CAP}-character cap`);
}
const rest = links.slice(linkStart);
let nLinks = count(rest, (lines.length ? Math.floor(QUERY_REPLY_CAP / 2) : QUERY_REPLY_CAP) - base);
if (!nLinks && rest.length) nLinks = count(rest, QUERY_REPLY_CAP - base);
if (!nLinks && rest.length) raise('query-item-too-large', `a link of ${file.path} alone is over the ${QUERY_REPLY_CAP}-character cap`);
while (lines.length && nLinks > 0 && bytes(page(1, nLinks)) > QUERY_REPLY_CAP) nLinks = Math.floor(nLinks / 2);
let nLines = Math.max(lines.length ? 1 : 0, count(lines, QUERY_REPLY_CAP - bytes(page(0, nLinks))));
while (nLines > 1 && bytes(page(nLines, nLinks)) > QUERY_REPLY_CAP) nLines -= 1;
return checked(page(nLines, nLinks));
}
// search: lines whose text holds `text`, case-insensitive. The excerpt is the whole line. Up to 50 per page.
export async function search(source, { universe: name, text, book, scope, offset } = {}) {
const universe = universeArg(source.mapping, name);
const needle = textArg(text, 'text').toLowerCase();
if (!needle) invalid('text must not be empty');
const { index, files } = await visible(source, universe, {
scope: scopeArg(scope), book: book === undefined ? undefined : textArg(book, 'book'),
});
const hits = [];
for (const file of files) {
for (const [line, body] of file.lines) {
if (!body.toLowerCase().includes(needle)) continue;
const hit = { path: file.path, line, excerpt: body, scope: file.scope };
// The excerpt is the whole line. A line too long to fit alone is marked, not cut and not refused.
hits.push(bytes({ commit: index.commit, results: [hit], next: Number.MAX_SAFE_INTEGER }) > QUERY_REPLY_CAP
? { path: file.path, line, excerpt: null, excerpt_too_long: true, scope: file.scope } : hit);
}
}
return fitPage(hits, {
offset: offsetArg(offset),
build: (picked, next) => ({ commit: index.commit, results: picked, next }),
});
}
// manuscript: one chapter file of a book, by its file name, up to 400 lines per call from `from` to `to`.
// Without a chapter it returns the book's first chapter file in path order. Every reply carries
// `next_chapter`, the next chapter file's name in path order or null. A name that matches two visible
// files is refused. Files outside every book (book null) are never chapters of a book.
export async function manuscript(source, { universe: name, book, scope, chapter, from, to } = {}) {
const universe = universeArg(source.mapping, name);
const bookName = textArg(book, 'book');
const chapterName = chapter === undefined ? undefined : textArg(chapter, 'chapter');
const start = positionArg(from, 'from', 1);
const end = positionArg(to, 'to', Infinity);
if (end < start) invalid('to must not be before from');
const { index, files } = await visible(source, universe, { scope: scopeArg(scope), book: bookName, type: 'manuscript' });
let at = 0;
if (chapterName !== undefined) {
const matches = files.filter(file => file.stem === chapterName);
if (matches.length > 1) raise('query-chapter-ambiguous', `${matches.length} visible files are named ${JSON.stringify(chapterName)}`);
if (!matches.length) return null;
at = files.indexOf(matches[0]);
}
const file = files[at];
if (!file) return null;
const nextChapter = files[at + 1]?.stem ?? null;
return fitPage(file.lines.slice(start - 1, end), {
limit: MANUSCRIPT_LINES,
build: (picked, next) => ({
path: file.path, commit: index.commit, scope: file.scope, lines: picked,
next: next === null ? null : start + next, next_chapter: nextChapter,
}),
});
}
// universeInfo: a universe's kind, name, books, playground prefixes, type folders and manuscript folders.
// A hybrid adds its sources, constraints and notes. A draft has no books and no canon: its kind, name
// and prefixes are the whole answer beside the mapping's types. It reads only the checked mapping and the
// served commit.
export async function universeInfo(source, { universe: name } = {}) {
const universe = universeArg(source.mapping, name);
const entry = source.mapping.universes[universe];
const hybrid = entry.kind === 'hybrid';
return checked({
commit: source.commit, universe, name: entry.name, kind: entry.kind ?? 'universe',
books: entry.books, playground: entry.playground, types: source.mapping.types, manuscript: source.mapping.manuscript,
...(hybrid ? {
sources: entry.sources, constraints: entry.constraints,
...(entry.notes === undefined ? {} : { notes: entry.notes }),
} : {}),
});
}
@@ -0,0 +1,91 @@
// Index tests (C2-05a): the cache key and its bound, failed reads, mapped-only contents, and the rebuild
// time on a 3,000-file fixture. Fixtures are made up and built under the scratch area.
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { symlinkSync, writeFileSync } from 'node:fs';
import { join } from 'node:path';
import { performance } from 'node:perf_hooks';
import { clearIndexCache, indexCacheSize, indexFor } from '../src/index-build.mjs';
import { git, mapping, MARKER } from './helpers.mjs';
import { note, scratchRepo } from './query-fixture.mjs';
const FILES = {
'Book One/Characters/Mara Vell.md': note('character', 'Mara Vell', MARKER),
'Book One/Chapters/Chapter 01.md': '# One\n[[Mara Vell]]\n',
'Inbox Note.md': `${MARKER}\n`,
};
// A mapping whose harbor universe has the given display name. Each name gives a new cache key.
const named = name => mapping(e => ({ ...e, universes: { ...e.universes, harbor: { ...e.universes.harbor, name } } }));
test('the same commit, repo path and mapping return the same index, and only mapped files are in it', async t => {
clearIndexCache();
const { repo, commit } = scratchRepo(t, FILES);
const a = await indexFor({ repo, commit, mapping: mapping() });
const b = await indexFor({ repo, commit, mapping: mapping() });
assert.equal(a, b);
assert.equal(indexCacheSize(), 1);
assert.equal(a.commit, commit);
assert.ok(a.files.has('Book One/Characters/Mara Vell.md'));
assert.ok(!a.files.has('Inbox Note.md'));
assert.deepEqual(a.names.get('mara vell'), ['Book One/Characters/Mara Vell.md']);
assert.deepEqual(a.linksIn.get('Book One/Characters/Mara Vell.md').map(l => l.path), ['Book One/Chapters/Chapter 01.md']);
});
test('a different repo path, mapping or commit gets its own index', async t => {
clearIndexCache();
const { root, repo, commit } = scratchRepo(t, FILES);
const base = await indexFor({ repo, commit, mapping: mapping() });
const alias = join(root, 'alias');
symlinkSync(repo, alias);
assert.notEqual(await indexFor({ repo: alias, commit, mapping: mapping() }), base);
assert.notEqual(await indexFor({ repo, commit, mapping: named('Renamed Harbor') }), base);
writeFileSync(join(repo, 'Book One/Places/New.md'), note('place', 'New'));
git(repo, 'add', '-A');
git(repo, 'commit', '-qm', 'second');
const next = git(repo, 'rev-parse', 'HEAD');
const later = await indexFor({ repo, commit: next, mapping: mapping() });
assert.notEqual(later, base);
assert.equal(later.commit, next);
assert.ok(later.files.has('Book One/Places/New.md'));
});
test('the cache holds at most eight indexes and evicts the least recently used', async t => {
clearIndexCache();
const { repo, commit } = scratchRepo(t, FILES);
const maps = Array.from({ length: 9 }, (_, i) => named(`Harbor ${i}`));
const first = await indexFor({ repo, commit, mapping: maps[0] });
for (const m of maps.slice(1)) await indexFor({ repo, commit, mapping: m });
assert.equal(indexCacheSize(), 8);
assert.notEqual(await indexFor({ repo, commit, mapping: maps[0] }), first);
assert.equal(indexCacheSize(), 8);
});
test('a failed read is not kept in the cache', async t => {
clearIndexCache();
const { repo } = scratchRepo(t, FILES);
await assert.rejects(indexFor({ repo, commit: '0'.repeat(40), mapping: mapping() }));
assert.equal(indexCacheSize(), 0);
});
test('a missing or unchecked mapping is refused before any read', async t => {
clearIndexCache();
const { repo, commit } = scratchRepo(t, FILES);
await assert.rejects(indexFor({ repo, commit, mapping: undefined }), { code: 'library-mapping-invalid' });
await assert.rejects(indexFor({ repo, commit, mapping: { kind: 'hybrid' } }), { code: 'library-mapping-invalid' });
});
test('a 3,000-file rebuild completes and its time is reported', { timeout: 300_000 }, async t => {
clearIndexCache();
const folders = ['Characters', 'Places', 'Research'];
const files = {};
for (let i = 0; i < 3000; i++) {
files[`Book One/${folders[i % 3]}/Note ${i}.md`] = note('character', `Note ${i}`, `Body ${i} [[Note ${(i + 1) % 3000}]]`);
}
const { repo, commit } = scratchRepo(t, files);
const started = performance.now();
const index = await indexFor({ repo, commit, mapping: mapping() });
const ms = performance.now() - started;
t.diagnostic(`3,000-file index rebuild: ${ms.toFixed(0)} ms`);
assert.equal(index.paths.length, 3000);
});
+31
View File
@@ -0,0 +1,31 @@
// Scratch repos for the query and index tests (C2-05a). They are built under tmpdir(), as helpers.mjs
// repo() does; the gate sets TMPDIR under /mnt/storage/scratch/tmp, never /tmp (RULINGS "tmpdir flag").
// Bodies are made up.
import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { dirname, join } from 'node:path';
import { git } from './helpers.mjs';
// A repo with one commit holding `files` (path -> text). Removed when the test ends.
export function scratchRepo(t, files) {
const root = mkdtempSync(join(tmpdir(), 'r216-'));
t.after(() => rmSync(root, { recursive: true, force: true }));
const repo = join(root, 'books');
mkdirSync(repo);
git(repo, 'init', '-q', '-b', 'main');
for (const [path, text] of Object.entries(files)) {
mkdirSync(dirname(join(repo, path)), { recursive: true });
writeFileSync(join(repo, path), text);
}
git(repo, 'add', '-A');
git(repo, 'commit', '-qm', 'fixture');
return { root, repo, commit: git(repo, 'rev-parse', 'HEAD') };
}
// A note: front matter with a type and a name, optional extra front matter lines, then a body.
export const note = (type, name, body = '', extra = []) => ['---', `type: ${type}`, `name: ${name}`, ...extra, '---', body, ''].join('\n');
// The one exclusion the hybrid tests use, and the paths it hides. C2-15 (row 267) replaces the reading of
// `exclude` (a path prefix today); swapping these two constants is the whole change to the tests.
export const EXCLUSION = { exclude: 'Book One/Research' };
export const EXCLUDED_PATHS = ['Book One/Research/Ferry Logs.md'];
+369
View File
@@ -0,0 +1,369 @@
// Query tests for design 3 (C2-05a): universe isolation, scope labels, unmapped files, hybrids, paging
// under the reply cap, refusals and manuscript paging. Fixtures are made up and built under the scratch area.
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { REPLY_CAP } from '../../harness/src/tools.mjs';
import { find, get, search, manuscript, universeInfo, QUERY_REPLY_CAP, LOOKUP_LIMIT, MANUSCRIPT_LINES } from '../src/query.mjs';
import { mapping, MARKER } from './helpers.mjs';
import { scratchRepo, note, EXCLUSION, EXCLUDED_PATHS } from './query-fixture.mjs';
// A hybrid over harbor and second. Its one constraint is EXCLUSION (query-fixture.mjs) and it declares no
// books or playground.
const WITH_BLEND = e => ({
...e,
universes: {
...e.universes,
blend: {
kind: 'hybrid', name: 'Blend', sources: ['harbor', 'second'], books: [], playground: [],
constraints: [EXCLUSION], notes: 'made-up hybrid',
},
},
});
// PRD 1.14 levels (C2-01c): harbor gets a universe-level playground prefix outside every book, and a draft
// universe has only playground prefixes. Files there have book null.
const WITH_LEVELS = e => ({
...WITH_BLEND(e),
universes: {
...WITH_BLEND(e).universes,
harbor: { ...e.universes.harbor, playground: ['Book One/Playground/', 'Notes/Harbor/'] },
idea: { kind: 'draft', name: 'New World', playground: ['Drafts/New World/'] },
},
});
const LEVEL_FILES = {
'Notes/Harbor/Scraps.md': note('summary', 'Scraps', `${MARKER} loose`),
'Notes/Harbor/Chapters/Idea Chapter.md': `# Idea\n${MARKER} unbooked chapter\n`,
'Drafts/New World/Characters/Rin.md': note('character', 'Rin', `${MARKER} draft`),
'Drafts/New World/Summary.md': note('summary', 'Summary', `${MARKER} draft`),
};
const FILES = {
'Book One/Characters/Mara Vell.md': note('character', 'Mara Vell', `Grey eyes. ${MARKER}`,
['appears_in: "[[Book One/Chapters/Chapter 01]]"', 'relationships:', ' - "[[Tomas]]"']),
'Book One/Characters/Tomas.md': note('character', 'Tomas', MARKER),
'Book One/Chapters/Chapter 01.md': `# One\n${MARKER} [[Mara Vell]] met [[Book One/Places/Harbor Town|the harbor]].\n`,
'Book One/Chapters/Chapter 02.md': 'Quiet. [[Ferry Logs]] [[Ilse]]\n',
'Book One/Places/Harbor Town.md': note('place', 'Harbor Town', `Salt. ${MARKER}`),
'Book One/Research/Ferry Logs.md': note('research', 'Ferry Logs', `Timetable. ${MARKER}`),
'Book One/Playground/What If.md': note('summary', 'What If', `${MARKER} [[Mara Vell]]`),
'Second World/Characters/Ilse.md': note('character', 'Ilse', `${MARKER} Ilse.`),
'Second World/Chapters/01.md': '# Elsewhere\n[[Ilse]] waits.\n',
'Inbox Note.md': `${MARKER}\n`,
'Book One/Board.canvas': '{"nodes":[]}\n',
};
const src = (ref, map = mapping()) => ({ repo: ref.repo, commit: ref.commit, mapping: map });
const paths = reply => reply.results.map(r => r.path);
test('the query reply cap equals the harness reply cap', () => {
assert.equal(QUERY_REPLY_CAP, REPLY_CAP);
});
test('find is bound to its universe and refuses a call without one', async t => {
const s = src(scratchRepo(t, FILES));
const names = paths(await find(s, { universe: 'harbor', query: '' }));
assert.ok(names.length > 0);
assert.ok(names.every(p => p.startsWith('Book One/')));
assert.ok(!names.includes('Second World/Characters/Ilse.md'));
await assert.rejects(find(s, { query: '' }), { code: 'query-argument-invalid' });
await assert.rejects(find(s, { universe: 'nowhere', query: '' }), { code: 'query-universe-unknown' });
});
test('every result carries its scope, and playground appears only when asked for', async t => {
const s = src(scratchRepo(t, FILES));
const canon = (await find(s, { universe: 'harbor', query: '' })).results;
assert.ok(canon.every(r => r.scope === 'canon'));
assert.ok(!canon.some(r => r.path.includes('/Playground/')));
const play = (await find(s, { universe: 'harbor', query: '', scope: 'playground' })).results;
assert.deepEqual(play.map(r => r.path), ['Book One/Playground/What If.md']);
assert.ok(play.every(r => r.scope === 'playground'));
});
test('unmapped files never appear in any reply', async t => {
const s = src(scratchRepo(t, FILES));
assert.deepEqual((await find(s, { universe: 'harbor', query: 'Inbox' })).results, []);
assert.deepEqual((await find(s, { universe: 'harbor', query: 'Board' })).results, []);
const hits = (await search(s, { universe: 'harbor', text: MARKER })).results;
assert.ok(hits.length > 0);
assert.ok(hits.every(r => r.path !== 'Inbox Note.md' && r.path !== 'Book One/Board.canvas'));
});
test('a hybrid sees its sources canon, minus exclusions, and no playground', async t => {
const s = src(scratchRepo(t, FILES), mapping(WITH_BLEND));
const reply = await find(s, { universe: 'blend', query: '' });
assert.deepEqual(paths(reply).sort(), [
'Book One/Characters/Mara Vell.md', 'Book One/Characters/Tomas.md', 'Book One/Chapters/Chapter 01.md',
'Book One/Chapters/Chapter 02.md', 'Book One/Places/Harbor Town.md',
'Second World/Characters/Ilse.md', 'Second World/Chapters/01.md',
]);
assert.ok(reply.results.every(r => r.scope === 'canon' && ['harbor', 'second'].includes(r.universe)));
assert.deepEqual((await find(s, { universe: 'blend', query: '', scope: 'playground' })).results, []);
assert.deepEqual((await find(s, { universe: 'blend', query: 'Ferry' })).results, []);
});
test('a hybrid gets a source file it can see, and not an excluded or foreign one', async t => {
const ref = scratchRepo(t, FILES);
const hybrid = src(ref, mapping(WITH_BLEND));
const ilse = await get(hybrid, { universe: 'blend', path: 'Second World/Characters/Ilse.md' });
assert.equal(ilse.scope, 'canon');
assert.ok(ilse.links_in.some(l => l.path === 'Second World/Chapters/01.md'));
assert.equal(await get(hybrid, { universe: 'blend', path: 'Book One/Research/Ferry Logs.md' }), null);
assert.equal(await get(src(ref), { universe: 'harbor', path: 'Second World/Characters/Ilse.md' }), null);
});
test('get lists visible links both ways and drops a target another universe cannot see', async t => {
const ref = scratchRepo(t, FILES);
const harbor = src(ref);
const ch1 = await get(harbor, { universe: 'harbor', path: 'Book One/Chapters/Chapter 01.md' });
assert.ok(ch1.links_out.some(l => l.path === 'Book One/Characters/Mara Vell.md' && l.line === 2));
assert.ok(ch1.links_out.some(l => l.path === 'Book One/Places/Harbor Town.md'));
const ch2 = await get(harbor, { universe: 'harbor', path: 'Book One/Chapters/Chapter 02.md' });
assert.ok(ch2.links_out.some(l => l.path === 'Book One/Research/Ferry Logs.md'));
assert.ok(!ch2.links_out.some(l => l.path === 'Second World/Characters/Ilse.md'));
const mara = await get(harbor, { universe: 'harbor', path: 'Book One/Characters/Mara Vell.md' });
assert.ok(mara.links_in.some(l => l.path === 'Book One/Chapters/Chapter 01.md'));
assert.ok(mara.links_in.some(l => l.path === 'Book One/Playground/What If.md'));
assert.ok(mara.links_out.some(l => l.path === 'Book One/Characters/Tomas.md' && l.field === 'relationships'));
assert.ok(mara.links_out.some(l => l.path === 'Book One/Chapters/Chapter 01.md' && l.field === 'appears_in'));
const blendCh2 = await get(src(ref, mapping(WITH_BLEND)), { universe: 'blend', path: 'Book One/Chapters/Chapter 02.md' });
assert.ok(blendCh2.links_out.some(l => l.path === 'Second World/Characters/Ilse.md'));
});
test('a long list pages by position, never cuts an item, and returns 50 per page', async t => {
const files = { ...FILES };
for (let i = 1; i <= 120; i++) files[`Book One/Places/Note ${i}.md`] = note('place', `Note ${i}`);
const s = src(scratchRepo(t, files));
const seen = [];
let offset = 0;
for (let pages = 0; pages < 10; pages++) {
const reply = await find(s, { universe: 'harbor', query: 'Note', offset });
assert.ok(JSON.stringify(reply).length <= QUERY_REPLY_CAP);
assert.ok(reply.results.length <= LOOKUP_LIMIT);
if (pages === 0) assert.equal(reply.results.length, LOOKUP_LIMIT);
seen.push(...paths(reply));
if (reply.next === null) break;
offset = reply.next;
}
assert.equal(seen.length, 120);
assert.equal(new Set(seen).size, 120);
});
test('a page stays under the reply cap when items are large', async t => {
const big = 'x'.repeat(2000);
const files = { ...FILES };
for (let i = 1; i <= 40; i++) files[`Book One/Places/Big ${i}.md`] = note('place', `Big ${i}`, '', [`appears_in: "${big}"`]);
const s = src(scratchRepo(t, files));
const first = await find(s, { universe: 'harbor', query: 'Big' });
assert.ok(JSON.stringify(first).length <= QUERY_REPLY_CAP);
assert.ok(first.results.length > 0 && first.results.length < 40);
assert.notEqual(first.next, null);
const seen = [];
let offset = 0;
for (let pages = 0; pages < 10; pages++) {
const reply = await find(s, { universe: 'harbor', query: 'Big', offset });
assert.ok(JSON.stringify(reply).length <= QUERY_REPLY_CAP);
seen.push(...paths(reply));
if (reply.next === null) break;
offset = reply.next;
}
assert.equal(new Set(seen).size, 40);
});
test('a search line too long to fit alone comes back marked, with no excerpt, and the other hits still return', async t => {
const huge = 'y'.repeat(QUERY_REPLY_CAP + 1000);
const s = src(scratchRepo(t, { ...FILES, 'Book One/Places/Wall.md': note('place', 'Wall', `${huge} ${MARKER}`) }));
const reply = await search(s, { universe: 'harbor', text: 'yyyyyyyy' });
assert.deepEqual(reply.results, [{ path: 'Book One/Places/Wall.md', line: 5, excerpt: null, excerpt_too_long: true, scope: 'canon' }]);
const mixed = await search(s, { universe: 'harbor', text: MARKER.toLowerCase() });
assert.ok(mixed.results.some(r => r.excerpt_too_long === true));
assert.ok(mixed.results.some(r => typeof r.excerpt === 'string' && !('excerpt_too_long' in r)));
assert.ok(JSON.stringify(mixed).length <= QUERY_REPLY_CAP);
});
test('a find item that cannot fit alone is refused, not cut', async t => {
const huge = 'y'.repeat(QUERY_REPLY_CAP + 1000);
const s = src(scratchRepo(t, { ...FILES, 'Book One/Places/Wall.md': note('place', 'Wall', '', [`appears_in: "${huge}"`]) }));
await assert.rejects(find(s, { universe: 'harbor', query: 'Wall' }), { code: 'query-item-too-large' });
});
test('get pages its lines, and the next line picks up where the page stopped', async t => {
const body = `${Array.from({ length: 500 }, (_, i) => `Line ${i + 1} ${'z'.repeat(120)}`).join('\n')}\n`;
const s = src(scratchRepo(t, { ...FILES, 'Book One/Chapters/Long.md': body }));
const first = await get(s, { universe: 'harbor', path: 'Book One/Chapters/Long.md' });
assert.equal(first.lines[0][0], 1);
assert.ok(first.lines.length < 501);
assert.notEqual(first.next, null);
let from = 1, count = 0;
for (let pages = 0; pages < 10; pages++) {
const page = await get(s, { universe: 'harbor', path: 'Book One/Chapters/Long.md', from });
count += page.lines.length;
if (page.next === null) break;
from = page.next;
}
assert.equal(count, 501);
});
test('get refuses a file whose front matter alone is over the cap, and a line alone is over the cap', async t => {
const huge = 'q'.repeat(QUERY_REPLY_CAP + 1000);
const s = src(scratchRepo(t, {
...FILES,
'Book One/Places/Heavy.md': note('place', 'Heavy', '', [`appears_in: "${huge}"`]),
'Book One/Chapters/Wide.md': `${huge}\n`,
}));
await assert.rejects(get(s, { universe: 'harbor', path: 'Book One/Places/Heavy.md' }), { code: 'query-item-too-large' });
await assert.rejects(get(s, { universe: 'harbor', path: 'Book One/Chapters/Wide.md' }), { code: 'query-item-too-large' });
});
test('get pages the links of a hub note, every page carries the front matter, and the pages add up', async t => {
const files = { ...FILES };
for (let i = 1; i <= 900; i++) files[`Book One/Chapters/Mention ${i}.md`] = `[[Tomas]] ${'w'.repeat(60)}\n`;
const s = src(scratchRepo(t, files));
const seen = [];
const PAST_END = 1000;
let linksFrom = 0, pages = 0, lines = 0, from = 1;
for (; pages < 40; pages++) {
const page = await get(s, { universe: 'harbor', path: 'Book One/Characters/Tomas.md', from, links_from: linksFrom });
assert.ok(JSON.stringify(page).length <= QUERY_REPLY_CAP);
assert.equal(page.frontmatter.name, 'Tomas');
seen.push(...page.links_in.map(l => l.path));
if (from !== PAST_END) lines += page.lines.length;
// Once the lines are read, ask past the last one: the page then carries links alone.
from = page.next ?? PAST_END;
if (page.links_next === null && page.next === null) break;
if (page.links_next !== null) linksFrom = page.links_next;
}
assert.ok(pages > 1, 'the hub needed more than one page');
assert.ok(pages < 40);
assert.equal(seen.filter(p => p.startsWith('Book One/Chapters/Mention ')).length, 900);
assert.equal(new Set(seen).size, seen.length, 'no link comes back twice');
assert.equal(lines, 6, 'the five lines of the note and the empty one after its last newline');
const tail = await get(s, { universe: 'harbor', path: 'Book One/Characters/Tomas.md', from: PAST_END });
assert.deepEqual(tail.lines, []);
assert.equal(tail.next, null);
});
test('get rejects a bad links_from', async t => {
const s = src(scratchRepo(t, FILES));
await assert.rejects(get(s, { universe: 'harbor', path: 'Book One/Characters/Tomas.md', links_from: -1 }), { code: 'query-argument-invalid' });
const past = await get(s, { universe: 'harbor', path: 'Book One/Characters/Tomas.md', links_from: 99 });
assert.deepEqual([past.links_out, past.links_in, past.links_next], [[], [], null]);
});
test('search returns scoped line hits, case-insensitively, and refuses empty text', async t => {
const s = src(scratchRepo(t, FILES));
const reply = await search(s, { universe: 'harbor', text: 'marker' });
assert.ok(reply.results.length > 0);
assert.ok(reply.results.every(r => r.scope === 'canon' && r.path.startsWith('Book One/') && Number.isInteger(r.line)));
assert.ok(reply.results.some(r => r.path === 'Book One/Chapters/Chapter 01.md' && r.line === 2));
await assert.rejects(search(s, { universe: 'harbor', text: '' }), { code: 'query-argument-invalid' });
});
test('a manuscript call returns one chapter, at most 400 lines, and names its next line', async t => {
const long = `${Array.from({ length: 900 }, (_, i) => `Scene ${i + 1}`).join('\n')}\n`;
const s = src(scratchRepo(t, { ...FILES, 'Book One/Chapters/Long Chapter.md': long }));
const first = await manuscript(s, { universe: 'harbor', book: 'Book One', chapter: 'Long Chapter' });
assert.equal(first.lines.length, MANUSCRIPT_LINES);
assert.equal(first.next, MANUSCRIPT_LINES + 1);
const second = await manuscript(s, { universe: 'harbor', book: 'Book One', chapter: 'Long Chapter', from: first.next });
assert.equal(second.lines[0][0], MANUSCRIPT_LINES + 1);
});
test('a manuscript call without a chapter returns the first chapter in path order and names the next', async t => {
const s = src(scratchRepo(t, FILES));
const first = await manuscript(s, { universe: 'harbor', book: 'Book One' });
assert.equal(first.path, 'Book One/Chapters/Chapter 01.md');
assert.equal(first.next_chapter, 'Chapter 02');
const last = await manuscript(s, { universe: 'harbor', book: 'Book One', chapter: 'Chapter 02' });
assert.equal(last.path, 'Book One/Chapters/Chapter 02.md');
assert.equal(last.next_chapter, null);
assert.equal(await manuscript(s, { universe: 'harbor', book: 'Book One', chapter: 'Nope' }), null);
});
test('a chapter name that matches two visible files is refused', async t => {
const s = src(scratchRepo(t, { ...FILES, 'Book One/Chapters/Part/Chapter 01.md': 'Again.\n' }));
await assert.rejects(manuscript(s, { universe: 'harbor', book: 'Book One', chapter: 'Chapter 01' }), { code: 'query-chapter-ambiguous' });
});
test('a hybrid reads a source chapter through its declared sources only', async t => {
const ref = scratchRepo(t, FILES);
const hybrid = await manuscript(src(ref, mapping(WITH_BLEND)), { universe: 'blend', book: 'Second World', chapter: '01' });
assert.equal(hybrid.path, 'Second World/Chapters/01.md');
assert.equal(hybrid.scope, 'canon');
assert.equal(await manuscript(src(ref), { universe: 'harbor', book: 'Second World', chapter: '01' }), null);
});
test('universe info reports a hybrid sources, constraints and notes, and refuses an unknown universe', async t => {
const s = src(scratchRepo(t, FILES), mapping(WITH_BLEND));
const blend = await universeInfo(s, { universe: 'blend' });
assert.equal(blend.kind, 'hybrid');
assert.deepEqual(blend.sources, ['harbor', 'second']);
assert.deepEqual(blend.constraints, [EXCLUSION]);
assert.equal(blend.notes, 'made-up hybrid');
const harbor = await universeInfo(s, { universe: 'harbor' });
assert.equal(harbor.kind, 'universe');
assert.equal(harbor.sources, undefined);
await assert.rejects(universeInfo(s, { universe: 'ghost' }), { code: 'query-universe-unknown' });
});
test('an exclusion hides what it names from a hybrid and from nothing else', async t => {
const ref = scratchRepo(t, FILES);
const hybrid = src(ref, mapping(WITH_BLEND));
const seen = paths(await find(hybrid, { universe: 'blend', query: '' }));
for (const path of EXCLUDED_PATHS) assert.ok(!seen.includes(path), `${path} stays hidden`);
const own = paths(await find(src(ref), { universe: 'harbor', query: '' }));
for (const path of EXCLUDED_PATHS) assert.ok(own.includes(path), `${path} stays visible to its own universe`);
});
test('PRD 1.14: a lookup that names a book leaves book-null files out', async t => {
const s = src(scratchRepo(t, { ...FILES, ...LEVEL_FILES }), mapping(WITH_LEVELS));
const all = paths(await find(s, { universe: 'harbor', query: '', scope: 'playground' }));
assert.deepEqual(all.sort(), ['Book One/Playground/What If.md', 'Notes/Harbor/Chapters/Idea Chapter.md', 'Notes/Harbor/Scraps.md']);
const booked = await find(s, { universe: 'harbor', query: '', scope: 'playground', book: 'Book One' });
assert.deepEqual(paths(booked), ['Book One/Playground/What If.md']);
assert.ok(booked.results.every(r => r.book === 'Book One'));
const hits = (await search(s, { universe: 'harbor', text: 'unbooked', scope: 'playground' })).results;
assert.deepEqual(hits.map(r => r.path), ['Notes/Harbor/Chapters/Idea Chapter.md']);
assert.deepEqual((await search(s, { universe: 'harbor', text: 'unbooked', scope: 'playground', book: 'Book One' })).results, []);
});
test('PRD 1.14: library_manuscript never reads a book-null file', async t => {
const s = src(scratchRepo(t, { ...FILES, ...LEVEL_FILES }), mapping(WITH_LEVELS));
// The unbooked file sits under a manuscript folder, so it has type manuscript and book null.
const listed = await find(s, { universe: 'harbor', query: 'Idea', scope: 'playground' });
assert.deepEqual(listed.results.map(r => [r.path, r.type, r.book]), [['Notes/Harbor/Chapters/Idea Chapter.md', 'manuscript', null]]);
assert.equal(await manuscript(s, { universe: 'harbor', book: 'Book One', scope: 'playground', chapter: 'Idea Chapter' }), null);
assert.equal(await manuscript(s, { universe: 'harbor', book: 'Book One', scope: 'playground' }), null);
assert.equal(await manuscript(s, { universe: 'harbor', book: 'Notes', scope: 'playground', chapter: 'Idea Chapter' }), null);
});
test('PRD 1.14: a draft universe has no canon, so its lookups return files only with scope playground', async t => {
const s = src(scratchRepo(t, { ...FILES, ...LEVEL_FILES }), mapping(WITH_LEVELS));
assert.deepEqual((await find(s, { universe: 'idea', query: '' })).results, []);
assert.deepEqual((await search(s, { universe: 'idea', text: MARKER })).results, []);
const found = await find(s, { universe: 'idea', query: '', scope: 'playground' });
assert.deepEqual(paths(found).sort(), ['Drafts/New World/Characters/Rin.md', 'Drafts/New World/Summary.md']);
assert.ok(found.results.every(r => r.scope === 'playground' && r.universe === 'idea' && r.book === null));
const hits = (await search(s, { universe: 'idea', text: 'draft', scope: 'playground' })).results;
assert.equal(hits.length, 2);
const rin = await get(s, { universe: 'idea', path: 'Drafts/New World/Characters/Rin.md' });
assert.equal(rin.scope, 'playground');
assert.equal(await get(s, { universe: 'harbor', path: 'Drafts/New World/Characters/Rin.md' }), null);
});
test('PRD 1.14: library_universe on a draft returns its kind, name and prefixes', async t => {
const s = src(scratchRepo(t, { ...FILES, ...LEVEL_FILES }), mapping(WITH_LEVELS));
const draft = await universeInfo(s, { universe: 'idea' });
assert.equal(draft.kind, 'draft');
assert.equal(draft.name, 'New World');
assert.deepEqual(draft.playground, ['Drafts/New World/']);
assert.deepEqual(draft.books, []);
assert.equal(draft.sources, undefined);
});
test('a hybrid gets a source playground file as null, while the source gets it with scope playground', async t => {
const ref = scratchRepo(t, FILES);
const path = 'Book One/Playground/What If.md';
assert.equal(await get(src(ref, mapping(WITH_BLEND)), { universe: 'blend', path }), null);
const own = await get(src(ref), { universe: 'harbor', path });
assert.equal(own.scope, 'playground');
});
-239
View File
@@ -1,239 +0,0 @@
// 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);
}
+4 -7
View File
@@ -97,12 +97,9 @@ 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
// 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 = [] }) {
// 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 = [] }) {
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" });
@@ -112,7 +109,7 @@ export function binds({ runDir, workspace, workspaceDest = null, adapter, harnes
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, ...(workspaceDest && workspaceDest !== workspace ? { dest: workspaceDest } : {}), why: "the instance's workspace, where the harness works" },
{ kind: "rw", src: workspace, 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" },
+17 -176
View File
@@ -20,22 +20,17 @@
// every launch, refusal and end
// workspaces/<business>/<instance>/ the instance's workspace (0700), kept
// across launches
// threads/<business>/<thread>/ a legacy thread session's store (0700),
// kept across its runs (row 121):
// threads/<business>/<thread>/ a thread's session store (0700), kept
// across the thread's 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, linkSync, lstatSync, mkdirSync, openSync, readdirSync, readFileSync, realpathSync, renameSync, unlinkSync, writeFileSync } from "node:fs";
import { appendFileSync, closeSync, existsSync, lstatSync, mkdirSync, openSync, readFileSync, renameSync, writeFileSync } from "node:fs";
import { constants } from "node:os";
import { dirname, join, sep } from "node:path";
import { dirname, join } from "node:path";
import { childrenOf, cmdlineOf, startTimeOf } from "./proc.mjs";
import { binds, prepareRun, REPO, sandboxArgv } from "./sandbox.mjs";
import { SeatError } from "./seat.mjs";
@@ -43,10 +38,7 @@ import { ID_PATTERN } from "../../business/src/vocabulary.mjs";
export { REPO };
export const RUNNER = join(REPO, "packages", "harness", "src", "runner.mjs");
// 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 SESSIONS_VERSION = 1;
export const sessionsFile = (dataRoot) => join(dataRoot, "bus-host", "sessions.json");
export const launchSocketPath = (dataRoot) => join(dataRoot, "bus-host", "launch.sock");
@@ -55,7 +47,6 @@ 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).
@@ -65,174 +56,24 @@ export const sessionWorkspaceDir = (dataRoot, business, instance, session) => jo
// 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("..");
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);
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);
}
}
// 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) {
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]) {
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 seatError(`${what}: ${d} is unsafe`, code);
if (st.isSymbolicLink() || !st.isDirectory() || st.uid !== process.getuid() || st.mode & 0o077) throw new SeatError(`thread store: ${d} is unsafe`, 2);
}
}
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).
@@ -371,10 +212,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 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;
// 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;
}
#write() {
if (this.#readOnly) return;
@@ -397,7 +238,7 @@ export function readSessions(dataRoot) {
} catch {
throw new SeatError(`sessions file is unreadable: ${file}`, 2);
}
if (!SESSIONS_READ.includes(doc?.sessionsVersion) || !Array.isArray(doc.sessions)) throw new SeatError(`sessions file is malformed: ${file}`, 2);
if (doc?.sessionsVersion !== SESSIONS_VERSION || !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 }));
}
-230
View File
@@ -1,230 +0,0 @@
// 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}`);
}
}
});
@@ -1,126 +0,0 @@
// 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]]);
});