Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/turn-tree-pooled-copy.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@schlessera/brain-ui-server": patch
---

A strict, voice or autonomous turn's data tree is built faster: admission copies its files and the index seed sixteen at a time off the event loop, requesting a reflink, instead of one file after another, and an aborted admission stops the copy and removes it. On the native fixture this halves the copy (2,000 notes 100 ms to 51 ms, 10,000 notes 471 ms to 245 ms).
18 changes: 18 additions & 0 deletions docs/decisions/governed-writable-workers.md
Original file line number Diff line number Diff line change
Expand Up @@ -852,6 +852,24 @@ configuration still loads from the frozen view at the brain path.
approved index copy. Applications write only non-hidden `.md` outside
`context/policies`, and the image-mask operation only its PNG and its
scratch prune's removals, so every other byte in the tree equals the view.
- **The copy runs as a pool (#2891).** Admission plans the tree (directories
and links) in one walk, then copies the files and the index seed sixteen at
a time off the event loop, requesting a reflink, and closes each directory's
mode and times last. Measured with the real admission code on a native
Btrfs checkpoint into the server's temporary directory, warm, as UID 1000
and as root (receipts on #2891, 2026-10-25): 2,000 notes of 16 KiB with a
115 MiB index 100 ms → 51 ms, 10,000 notes with a 574 MiB index
471 ms → 245 ms; a cold first copy 283 ms → 59 ms and 1,234 ms → 247 ms.
Two cheaper constructions were measured and not adopted. A reflink
placement on the managed volume takes 43 ms and 202 ms with one file after
another, because Bun's copy reflinks there already, but the server owns no
directory on that volume outside the live brain (the manager requires a
root-owned store, and the live subvolume is what governed peers write and
checkpoints qualify), so it needs a manager-provisioned location and a
capacity ruling first. Hardlinks from a shared base are refused by the
readers that require one link (`checkGovernancePolicy`, the embedding
policy, storage-attachment facts) and cannot cross subvolumes. The tree
still holds one copy per concurrent turn on the temporary directory.
- **Committed bytes enter the tree inside the commit boundary.** Each
application hands the tree its committed effects with the live base each
one replaced, partial staged failures included; an archive writes the
Expand Down
5 changes: 3 additions & 2 deletions docs/hosting/agent-workers.md
Original file line number Diff line number Diff line change
Expand Up @@ -401,8 +401,9 @@ write verified scratch/workspace copies through the aliases below.
commands that name the brain path see the turn-start bytes; a Claude `Read`
of a path the turn changed is refused and points to `mcp__brain__brain_read`, and
Claude's file tools do not read the tree itself.
Admission copies the startup view's documents into it (O(brain): 0.1-0.3 s
for 2,000 notes) and seeds its index, on the server's temporary directory.
Admission copies the startup view's documents into it (O(brain): about
0.05 s for 2,000 notes and 0.25 s for 10,000 on the native fixture) and
seeds its index, on the server's temporary directory.
Under a provider rule the tree leaves out the masked folders, and only
changes that pass the provenance rule enter it; the turn's private index is
read-only to the agent.
Expand Down
2 changes: 1 addition & 1 deletion packages/ui-server/src/brain/governed-turn.ts
Original file line number Diff line number Diff line change
Expand Up @@ -287,7 +287,7 @@ export function createGovernedTurns(options: {
// this turn's own applications (#2848); a masked one keeps its private
// index current from it (#2846). Built from this final checkpoint.
if (workerProfile(resolved) !== "governed" || privateIndex) {
dataTree = createTurnDataTree({ brainPath: options.brainPath, view: startupView!, maskPaths: options.maskPaths, mount: workerProfile(resolved) !== "governed",
dataTree = await createTurnDataTree({ brainPath: options.brainPath, view: startupView!, maskPaths: options.maskPaths, mount: workerProfile(resolved) !== "governed", signal,
...(privateIndex ? { index: privateIndex, masks: { folders: maskedFolders, buildFolders: privateIndexBuildFolders(maskedFolders, startupView!.root),
types: providerDeniedTypes(policy.policy, recipient) } } : {}),
// A recipient allowed every folder still never sees the generated-text caches.
Expand Down
87 changes: 61 additions & 26 deletions packages/ui-server/src/brain/turn-data-tree.ts
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,9 @@
* the next update drops it from the index too. A governed masked turn reads
* the live brain, so its tree feeds only its index updates.
*/
import { chmodSync, chownSync, closeSync, constants, copyFileSync, fstatSync, lstatSync, mkdirSync, mkdtempSync, openSync, readdirSync,
import { chmodSync, chownSync, closeSync, constants, fstatSync, lstatSync, mkdirSync, mkdtempSync, openSync, readdirSync,
readFileSync, readlinkSync, renameSync, rmSync, symlinkSync, unlinkSync, utimesSync, writeFileSync } from "node:fs";
import { chmod, copyFile, utimes } from "node:fs/promises";
import { tmpdir } from "node:os";
import { dirname, join, normalize, relative } from "node:path";
import { Database, constants as sqlite } from "bun:sqlite";
Expand Down Expand Up @@ -90,15 +91,43 @@ const COPY_ENTRIES = 200_000, COPY_DEPTH = 256;
const errno = (error: unknown) => (error as NodeJS.ErrnoException)?.code;

/**
* Copy the view's documents into `target`: links stay links, timestamps and
* modes are kept, nothing else (sockets, devices) is copied. `skip` names
* repo-relative paths left out. Across UIDs, a file or directory the server
* cannot read (a worker's own `0600` file) is left out too: the worker that
* could read it sees it at the brain path.
* Files copied at once. A copy onto the server's temporary directory crosses
* filesystems and waits on I/O, so a pool takes about half the time of one
* file after another (#2891); a reflink, where the filesystem offers one, is
* requested and taken.
*/
function copyView(view: string, target: string, skip: (rel: string) => boolean): void {
const COPY_WIDTH = 16;
const COPY_FLAGS = constants.COPYFILE_EXCL | constants.COPYFILE_FICLONE;

/**
* Run `jobs` at most `COPY_WIDTH` at a time. The first failure, or an abort,
* stops new jobs; it is thrown once every started job has settled, so nothing
* still writes into a tree its caller is about to remove.
*/
async function pooled(jobs: readonly (() => Promise<void>)[], signal?: AbortSignal): Promise<void> {
let next = 0, failed = false, failure: unknown;
await Promise.all(Array.from({ length: Math.min(COPY_WIDTH, jobs.length) }, async () => {
while (!failed && next < jobs.length) {
const job = jobs[next++]!;
try { signal?.throwIfAborted(); await job(); } catch (error) { if (!failed) { failed = true; failure = error; } }
}
}));
if (failed) throw failure;
}

/**
* Plan the copy of the view's documents into `target`: links stay links,
* timestamps and modes are kept, nothing else (sockets, devices) is copied.
* `skip` names repo-relative paths left out. Across UIDs, a file or directory
* the server cannot read (a worker's own `0600` file) is left out too: the
* worker that could read it sees it at the brain path. Directories and links
* are made here; the returned jobs copy the files, and `finish` then gives
* each directory its mode and times, deepest first.
*/
function planCopy(view: string, target: string, skip: (rel: string) => boolean): { files: (() => Promise<void>)[]; finish(): void } {
let entries = 0;
const unreadable = (error: unknown) => errno(error) === "EACCES";
const files: (() => Promise<void>)[] = [], directories: { to: string; info: import("node:fs").Stats }[] = [];
const walk = (rel: string, depth: number) => {
if (depth > COPY_DEPTH) throw new Error("The startup view exceeds the data tree's directory depth bound.");
let listed: import("node:fs").Dirent[];
Expand All @@ -109,28 +138,28 @@ function copyView(view: string, target: string, skip: (rel: string) => boolean):
if (++entries > COPY_ENTRIES) throw new Error(`The startup view exceeds the data tree's ${COPY_ENTRIES} entry bound.`);
const from = join(view, path), to = join(target, path), info = lstatSync(from);
if (info.isSymbolicLink()) symlinkSync(readlinkSync(from), to);
else if (info.isDirectory()) {
mkdirSync(to, { mode: 0o700 }); walk(path, depth + 1);
chmodSync(to, info.mode & 0o7777); utimesSync(to, info.atime, info.mtime);
} else if (info.isFile()) {
try { copyFileSync(from, to, constants.COPYFILE_EXCL); } catch (error) { if (unreadable(error)) continue; throw error; }
chmodSync(to, info.mode & 0o7777); utimesSync(to, info.atime, info.mtime);
}
else if (info.isDirectory()) { mkdirSync(to, { mode: 0o700 }); directories.push({ to, info }); walk(path, depth + 1); }
else if (info.isFile()) files.push(async () => {
try { await copyFile(from, to, COPY_FLAGS); } catch (error) { if (unreadable(error)) return; throw error; }
await chmod(to, info.mode & 0o7777); await utimes(to, info.atime, info.mtime);
});
}
};
walk("", 0);
// A directory is planned before everything in it, so the reverse order
// closes each one after its contents.
return { files, finish() { for (const { to, info } of directories.reverse()) { chmodSync(to, info.mode & 0o7777); utimesSync(to, info.atime, info.mtime); } } };
}

/** Seed `index` with the view's index files. The updater runs as the worker identity, which shares the server's group. */
function seedIndex(view: string, index: string, separateIdentity: boolean): void {
/** Jobs that seed `index` with the view's index files. The updater runs as the worker identity, which shares the server's group. */
function seedIndex(view: string, index: string, separateIdentity: boolean): (() => Promise<void>)[] {
const source = join(view, INDEX_DIR);
const names = (() => { try { return readdirSync(source, { withFileTypes: true }); } catch (error) { if (errno(error) === "ENOENT") return []; throw error; } })();
for (const entry of names) {
if (!entry.isFile()) continue;
return names.filter(entry => entry.isFile()).map(entry => async () => {
const to = join(index, entry.name);
copyFileSync(join(source, entry.name), to, constants.COPYFILE_EXCL);
chmodSync(to, separateIdentity ? 0o660 : 0o600);
}
await copyFile(join(source, entry.name), to, COPY_FLAGS);
await chmod(to, separateIdentity ? 0o660 : 0o600);
});
}

/**
Expand Down Expand Up @@ -278,9 +307,11 @@ async function runOnTree(options: {
* Copy the view into a new tree, before the worker launches. `index` is the
* masked turn's private index, already built; an unmasked turn's is seeded
* from the view. `pending` names paths the view's index may lag its documents
* on (`IndexBatch.watermark`): the first index read updates them.
* on (`IndexBatch.watermark`): the first index read updates them. The files
* and the index seed are copied together, off the event loop; an abort stops
* the copy and removes what it made.
*/
export function createTurnDataTree(options: {
export async function createTurnDataTree(options: {
brainPath: string; view: WorkerStartupView; maskPaths: readonly string[];
/** Mount it for the worker: false for a governed turn, which reads the live brain. */
mount: boolean;
Expand All @@ -295,7 +326,9 @@ export function createTurnDataTree(options: {
update?: (tree: WorkerDataTree, paths: readonly string[]) => Promise<void>;
/** Default: the maintenance program's capture planning, on the tree. */
plan?: (tree: WorkerDataTree, input: Record<string, unknown>, signal: AbortSignal) => Promise<unknown>;
}): TurnDataTree {
/** The admission's: stops the copy. */
signal?: AbortSignal;
}): Promise<TurnDataTree> {
const separateIdentity = Boolean(workerIdentityHelper());
const root = mkdtempSync(join(tmpdir(), "brain-turn-data-"));
let index: string | undefined;
Expand All @@ -307,14 +340,16 @@ export function createTurnDataTree(options: {
// The `.env*` masks cover link targets at the brain path only, so the tree
// leaves those targets out as well.
const hidden = [...(masks?.buildFolders ?? []), ...envMaskedPaths(options.view.root, options.brainPath)];
copyView(options.view.root, root, rel => {
const copy = planCopy(options.view.root, root, rel => {
const top = !rel.includes("/");
if (top && (OMITTED.has(rel) || isEnvName(rel) || (caches && PROVIDER_CACHE_FILES.includes(rel)) || (masks && rel === PROVIDER_CACHE_DIR))) return true;
return hidden.some(folder => rel === folder || rel.startsWith(`${folder}/`));
});
mkdirSync(join(root, INDEX_DIR), { mode: 0o755 });
index = options.index ?? createPrivateIndexDirectory();
if (!options.index) seedIndex(options.view.root, index, separateIdentity);
// The index files are the largest; they start first.
await pooled([...options.index ? [] : seedIndex(options.view.root, index, separateIdentity), ...copy.files], options.signal);
copy.finish();
settleIndex(index, separateIdentity);
} catch (error) {
try { removeCopy(root); } finally { if (index && !options.index) removePrivateIndexDirectory(index); }
Expand Down
2 changes: 1 addition & 1 deletion packages/ui-server/src/inbox/autonomous-turn.ts
Original file line number Diff line number Diff line change
Expand Up @@ -394,7 +394,7 @@ async function executeAutonomousTurn(
controller.signal.throwIfAborted();
workerProviderMasks = Object.freeze({ folders: Object.freeze([...maskedFolders]), index: privateIndex });
}
dataTree = createTurnDataTree({ brainPath, view: startupView, maskPaths: deps.privateIndex?.maskPaths ?? [], mount: true,
dataTree = await createTurnDataTree({ brainPath, view: startupView, maskPaths: deps.privateIndex?.maskPaths ?? [], mount: true, signal: controller.signal,
...(privateIndex ? { index: privateIndex, masks: { folders: maskedFolders, buildFolders: privateIndexBuildFolders(maskedFolders, startupView.root),
types: providerDeniedTypes(policy.policy, undefined) } } : {}),
pending: watermark?.paths() ?? [] });
Expand Down
4 changes: 2 additions & 2 deletions packages/ui-server/tests/private-index-stdio-release.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -105,7 +105,7 @@ for (const mode of ["success", "failure"] as const) {
const f = fixture();
const answer = mode === "success" ? { ok: true, plan: "Ithaca" } : { ok: false, error: "Penelope" };
const seen = observeLaunch({ pool: "input", script: `IFS= read -r request; sleep 0.3; printf '%s\\n' '${JSON.stringify(answer)}'` });
const tree = createTurnDataTree({ brainPath: f.brain, view: f.startupView, maskPaths: [], mount: true });
const tree = await createTurnDataTree({ brainPath: f.brain, view: f.startupView, maskPaths: [], mount: true });
cleanup.push(() => tree.close());
const result = await outcome(tree.planAdd({ title: "Ithaca" }, new AbortController().signal));
const open = seen.open();
Expand All @@ -124,7 +124,7 @@ for (const path of ["masked CLI", "turn tree"] as const) for (const abort of [fa
let child: Launched | undefined, settled = false;
observeLaunch({ script: path === "masked CLI" ? "printf Ithaca" : 'IFS= read -r request; printf \'{"ok":true,"plan":"Ithaca"}\\n\'',
withheld: { promise: release.promise, called: calling.resolve }, launched: launched => { child = launched; } });
const tree = path === "turn tree" ? createTurnDataTree({ brainPath: f.brain, view: f.startupView, maskPaths: [], mount: true }) : undefined;
const tree = path === "turn tree" ? await createTurnDataTree({ brainPath: f.brain, view: f.startupView, maskPaths: [], mount: true }) : undefined;
if (tree) cleanup.push(() => tree.close());
const work = outcome(tree ? tree.planAdd({}, controller.signal)
: runMaskedBrainCli({ ...f, signal: controller.signal, command: [] })).then(result => { settled = true; return result; });
Expand Down
Loading
Loading