fix(ui-server): commit the writer gate on the write worker

.changeset/writer-gate-off-event-loop.md

@@ -0,0 +1,5 @@
+---
+"@schlessera/brain-ui-server": patch
+---
+
+Changed: the hosted server commits the brain's writer gate on its SQLite write worker at `synchronous=FULL` instead of on the event loop (#2850). Each gate transaction keeps its `BEGIN IMMEDIATE`, so it stays serialized against the cron runner's. A release is submitted at once and settles once it has committed; a host process spawns only after its downgrade has committed and the boundary is still the host fallback; a type folder translation that cannot be recorded refuses the turn.

docs/decisions/governed-writable-workers.md

@@ -530,7 +530,7 @@ maintenance worker no longer receives the keys the provider relay carries
 (`GEMINI_API_KEY`, `ANTHROPIC_API_KEY`, `OPENAI_API_KEY` and keys the operator
 lists for it), and the brain's `.env*` files are masked as they are for a
 credential-free worker. Its `brain` CLI reaches those providers through its own
-relay (`providerRelayFor`, `packages/ui-server/src/brain/host-boundary.ts:417-424`).
+relay (`providerRelayFor`, `packages/ui-server/src/brain/host-boundary.ts:420-427`).
 What remains is narrower and stated in the status: keys the relay does not
 carry (the `jev` reranker's, a `brain sync` agent runner's login) and every
 key on a host without managed native storage, where there is no relay.
@@ -2286,7 +2286,7 @@ A peer whose plan only lacks a root is waited for, up to
 are planned only once a turn has translated it under the current policy hash,
 so writers admitted before that translation predate them, and every later
 writer inherits them from the held plan
-(`waitForPeerFolderPlans`, `packages/ui-server/src/brain/writer-gate.ts:261-272`).
+(`waitForPeerFolderPlans`, `packages/ui-server/src/brain/writer-gate.ts:342-353`).
 The translation is recorded in the UI database, so after the first one under a
 policy hash, writers in every process, the cron runner's and a restarted
 server's included, plan its folders without waiting (#2836).
@@ -2921,7 +2921,7 @@ index refresh or a cron job, retains them and the next turn's peer check
 passes. Admission records that translation in the UI database, one row per
 canonical brain holding the exact policy hash and its folders, so it reaches
 the cron runner's process and survives a restart (#2836;
-`function recordedTypeFolders`, `packages/ui-server/src/brain/writer-gate.ts:161-170`).
+`function recordedTypeFolders`, `packages/ui-server/src/brain/writer-gate.ts:173-182`).
 The host never translates it itself: only admission does, under its worker
 boundary. A missing, unreadable or other-hash record plans no type folder, and
 a stale one only retains an extra folder. Admission still refuses a peer
@@ -3672,7 +3672,7 @@ files, `.claude/`, `.agents/`, `.pi/`, `.mcp.json`, and
 - The core CLI loads `brain.config.ts` or `brain.config.json` at the canonical
   brain cwd (`loadUserConfig`, `packages/core/src/lib/config.ts:594-628`).
   Host commands bind that cwd to the selected brain (`function hostSpawn`,
-  `packages/ui-server/src/brain/host-boundary.ts:502-513`).
+  `packages/ui-server/src/brain/host-boundary.ts:511-522`).
 - Bun startup uses root `bunfig.toml` and `package.json`; retained literal
   preload/module inputs can identify a nested module's nearest package scope.
   The inventory walks parent directories only within the originating journal

docs/decisions/sqlite-write-worker.md

@@ -150,8 +150,22 @@ thread waits for it.
   invalidation still refuses the input that needed it; a failure retained
   before its session's row has committed is stored once it has, unless an
   input invalidated it meanwhile.
-- **Other app-connection writers stay on the loop for now**: the writer gate,
-  retained retries and local exchanges, saved worker profiles, policy
+- **The writer gate followed** (#2850). Acquisition, the turn-waiter queue,
+  release, folder retention, peer-plan checks, the unpinned-host downgrade,
+  the ungated host check and recorded type folders run their transactions on
+  the worker (`gateProcedures`, `packages/ui-server/src/brain/writer-gate.ts`),
+  each still `BEGIN IMMEDIATE`, so they stay serialized against the cron
+  runner's gate, which commits on its own connection. One policy reader on
+  the worker serves every observation the gate makes. A release is submitted
+  at once, so every later gate write of the process sees it, and settles once
+  it has committed; whatever promises the hold is gone (`withGate`, a turn's
+  release, a launch's `settled`, shutdown) waits for it. A host process
+  spawns only after its downgrade has committed, and only if the boundary is
+  still the host fallback afterwards. A type folder translation that cannot
+  be recorded refuses the turn. `holders()`, a diagnostic, still reads on
+  the gate's own connection.
+- **Other app-connection writers stay on the loop for now**: retained
+  retries and local exchanges, saved worker profiles, policy
   proposals, the Activity notifier and push ticks, auth sessions, inbox and
   settings. Each commits at
   FULL on the loop, and while a worker procedure holds its turn, such a write

docs/plans/storage-lifecycle.md

@@ -200,7 +200,7 @@ read the original receipt. Undo is a fresh stale-safe operation, not a blind cop
 over a newer file. Such a receipt protocol is proposed, not implemented here.
 
 The writer gate is cross-process and tracks process identity, not age
-(`export function createWriterGate`, `packages/ui-server/src/brain/writer-gate.ts:179-332`).
+(`export function gateProcedures`, `packages/ui-server/src/brain/writer-gate.ts:198-289`).
 Use that primitive alongside lifecycle ownership; do not interpret stale leases
 as proof an actually running backend is dead. Locks, retained readers and backups
 can deliberately prevent immediate reclamation.

packages/ui-server/src/app.ts

@@ -453,7 +453,7 @@ export async function createApp(options: CreateAppOptions = {}): Promise<BrainUi
   // It owns the journal's own file too (#2789), attached as it is here.
   const writer = openSqliteWriter(db, { attach: [{ schema: JOURNAL_SCHEMA, path: journalDatabasePath(config.dbPath) }] });
   bootCleanup.push(() => writer.close());
-  reconciled(hostExecution.attach(db));
+  reconciled(hostExecution.attach(db, writer));
   // Shared reference source (#2429) runs only where the native manager owns a reference pool.
   const referencePoolConfigured = await nativeManager.poolStatus().then(() => true, () => false);
   const referenceSources = referencePoolConfigured ? createReferenceSources({ db, pool: nativeManager, brain: nativeReady.brainUuid, credentials: referenceCredentialValues(),
@@ -1217,6 +1217,8 @@ export async function createApp(options: CreateAppOptions = {}): Promise<BrainUi
       await attempt(async () => { await referenceSources?.close(); });
       await attempt(() => nativeLeases.close());
       if (ephemeralJournal) await attempt(() => { rmSync(journalRoot, { recursive: true, force: true }); rmSync(storageCacheRoot, { recursive: true, force: true }); });
+      // Gate releases commit off the event loop (#2850): every one lands first.
+      await attempt(async () => { await hostExecution.gate()?.settled(); });
       await attempt(() => activity.close());
       await attempt(() => hostExecution.release());
       await attempt(() => db.close());

packages/ui-server/src/brain/governed-turn.ts

@@ -152,7 +152,7 @@ export function createGovernedTurns(options: {
       const discardPrivateIndex = () => { if (privateIndex) { const dir = privateIndex; privateIndex = undefined; removePrivateIndexDirectory(dir); } };
       const release = async () => {
         try { await dataTree?.close(); } finally {
-          try { await startupView?.cleanup(); } finally { try { await indexView?.cleanup(); } finally { try { discardPrivateIndex(); } finally { hold.release(); } } }
+          try { await startupView?.cleanup(); } finally { try { await indexView?.cleanup(); } finally { try { discardPrivateIndex(); } finally { await hold.release(); } } }
         }
       };
       // Marked before the checkpoint: its index may lack what is queued (#2848).
@@ -185,7 +185,7 @@ export function createGovernedTurns(options: {
         const typeFolderList = [...new Set(Object.values(typeFolders).flat())].sort();
         // Writers admitted later, by any process and after a restart, plan
         // these folders while the policy is unchanged (#2836).
-        if (types.length) gate.rememberTypeFolders(policy.hash, typeFolderList);
+        if (types.length) await gate.rememberTypeFolders(policy.hash, typeFolderList);
         const providerFolders = providerMaskedFolders(policy.policy, typeFolders);
         const maskedFolders = providerDeniedFolders(policy.policy, typeFolders, recipient);
         // A conversation's provider history keeps what its earlier turns, or a
@@ -209,7 +209,7 @@ export function createGovernedTurns(options: {
         // what its source kept masked.
         const lineage = !continued.length ? (maskedFolders.length ? maskedFolders : undefined)
           : lineages.every(always => always !== undefined) ? lineages.reduce<string[]>((kept, always) => maskedInBoth(kept, always!), maskedFolders) : undefined;
-        const folderPaths = hold.retainFolders?.(retainedFolderRoots(policy.policy, typeFolderList))
+        const folderPaths = await hold.retainFolders?.(retainedFolderRoots(policy.policy, typeFolderList))
           ?? [...new Set([...(hold.folderPaths ?? []), ...retainedFolderRoots(policy.policy, typeFolderList)])];
         // Retention needs the directory to exist; the host creates the index
         // and cache directories (never a provider folder) before the layout
@@ -250,7 +250,7 @@ export function createGovernedTurns(options: {
           // one started with the server before any turn had translated a type
           // selector, finishes rather than refusing the turn (#2832).
           if (hold.waitForPeerFolderPlans) await hold.waitForPeerFolderPlans(folderPaths, { signal, ...(options.peerFolderPlanWaitMs !== undefined ? { timeoutMs: options.peerFolderPlanWaitMs } : {}) });
-          else hold.assertPeerFolderPlans(folderPaths);
+          else await hold.assertPeerFolderPlans(folderPaths);
         }
         const commandPolicy = Object.freeze({ hash: policy.hash,
           outwardAllow: Object.freeze(policy.policy.outward.allow.map((rule, index) => Object.freeze({ ...rule, remotes: Object.freeze([...rule.remotes]), index: policy.outwardRuleIndices[index]! }))),

packages/ui-server/src/brain/host-boundary.ts

@@ -208,8 +208,11 @@ export interface HostBoundary {
   }): Promise<ExecutionResult>;
   /** The same, with live output, for a streamed sync or a cron job. */
   launch(execution: BrainExecution): Promise<StreamingExecution>;
-  /** A short synchronous query (git `check-ignore`). Takes no gate. */
-  runSync(execution: Omit<BrainExecution, "gate" | "signal" | "stdin"> & { timeoutMs: number; startupView?: WorkerStartupView }): ExecutionResult;
+  /**
+   * A short query run with a synchronous spawn (git `check-ignore`). Takes no
+   * gate; settles after the host fallback check commits off the event loop (#2850).
+   */
+  runSync(execution: Omit<BrainExecution, "gate" | "signal" | "stdin"> & { timeoutMs: number; startupView?: WorkerStartupView }): Promise<ExecutionResult>;
   /** Runs `fn` while holding the gate, when there is one. */
   withGate<T>(mode: GateMode, purpose: string, fn: () => Promise<T>, signal?: AbortSignal): Promise<T>;
 }
@@ -311,7 +314,7 @@ function credentialOutput(stream: ReadableStream<Uint8Array>, kill: () => void)
 
 /** Process exit alone need not mean its descendants released inherited output descriptors. */
 function retainExecutionAdmission(execution: StreamingExecution,
-  hold: { release(): void } | undefined): StreamingExecution {
+  hold: { release(): void | Promise<void> } | undefined): StreamingExecution {
   if (!hold) return execution;
   const kill = () => execution.kill("SIGKILL");
   const stdout = credentialOutput(execution.stdout, kill), stderr = credentialOutput(execution.stderr, kill);
@@ -474,7 +477,7 @@ export function createHostBoundary(options: HostBoundaryOptions): HostBoundary {
   }
   async function withGate<T>(mode: GateMode, purpose: string, fn: (hold: GateHold | null) => Promise<T>, signal?: AbortSignal): Promise<T> {
     const hold = await holdGate(mode, purpose, signal);
-    try { return await fn(hold); } finally { hold?.release(); }
+    try { return await fn(hold); } finally { await hold?.release(); }
   }
   const gated = <T>(execution: BrainExecution, fn: (hold: GateHold | null) => Promise<T>) =>
     execution.gate === "none" ? fn(null) : withGate(execution.gate, execution.purpose, fn, execution.signal);
@@ -492,10 +495,16 @@ export function createHostBoundary(options: HostBoundaryOptions): HostBoundary {
   // one (`gate: none`, runSync) still checks every active writer's retained
   // plan when a gate exists; with no gate at all, only the policy is known.
   const policyReader = createGovernancePolicyReader(brain);
-  const admitHost = (hold: GateHold | null) => {
+  // Each settles once its check has committed (#2850); nothing spawns before.
+  const admitHost = async (hold: GateHold | null) => {
     if (hold) return markUnpinnedHostWriter(hold);
     const gate = options.gate?.();
-    if (gate) gate.assertUngatedHostFallback(); else refuseHostFallback(policyReader.load());
+    if (gate) await gate.assertUngatedHostFallback(); else refuseHostFallback(policyReader.load());
+  };
+  /** After a host writer's admission commits: what was checked before it still holds, or nothing spawns. */
+  const stillHostFallback = (signal?: AbortSignal) => {
+    signal?.throwIfAborted();
+    if (isolated()) throw new Error("The brain's worker boundary changed while this host writer was being admitted; nothing ran. Retry.");
   };
 
   /** The pre-boundary path: a host process with the brain as its working directory. */
@@ -587,7 +596,9 @@ export function createHostBoundary(options: HostBoundaryOptions): HostBoundary {
             if (isolated()) return undefined;
             if (execution.outputBytes !== undefined || execution.publish === false) throw Error("Bounded Storage maintenance requires the qualified worker boundary");
             execution.assertAuthorized?.();
-            admitHost(hold);
+            await admitHost(hold);
+            stillHostFallback(admissionSignal);
+            execution.assertAuthorized?.();
             execution.onExecute?.();
             const proc = hostSpawn(execution);
             const completion = [new Response(proc.stdout).text(), new Response(proc.stderr).text(), proc.exited] as const;
@@ -652,11 +663,13 @@ export function createHostBoundary(options: HostBoundaryOptions): HostBoundary {
           if (isolated()) { credential?.release(); credential = undefined; }
           else {
             if (execution.publish === false) throw Error("A discarded read requires the qualified worker boundary");
-            admitHost(hold);
+            await admitHost(hold);
+            stillHostFallback(execution.signal);
             const proc = hostSpawn(execution);
+            const admitted = credential;
             return retainExecutionAdmission({ stdout: proc.stdout, stderr: proc.stderr, exited: proc.exited,
               get signalCode() { return proc.signalCode; }, kill: (signal?: NodeJS.Signals) => killWrapped(proc, exec, signal ?? "SIGTERM") },
-              credential || hold ? { release() { try { credential?.release(); } finally { hold?.release(); } } } : undefined);
+              admitted || hold ? { async release() { try { admitted?.release(); } finally { await hold?.release(); } } } : undefined);
           }
         } catch (error) { try { credential?.release(); } finally { hold?.release(); } throw error; }
       }
@@ -684,10 +697,10 @@ export function createHostBoundary(options: HostBoundaryOptions): HostBoundary {
             ...(execution.timeoutMs ? { timeoutMs: execution.timeoutMs } : {}) }).then(result => {
             stdout.write(result.stdout); stderr.write(result.stderr); return result.exitCode;
           }, error => { stderr.write(error instanceof Error ? error.message : String(error)); return 1; })
-            .finally(() => {
+            .finally(async () => {
               try { stdout.close(); } finally {
                 try { stderr.close(); } finally {
-                  try { execution.signal?.removeEventListener("abort", onAbort); } finally { try { releaseOutput(); } finally { hold?.release(); } }
+                  try { execution.signal?.removeEventListener("abort", onAbort); } finally { try { releaseOutput(); } finally { await hold?.release(); } }
                 }
               }
             });
@@ -729,19 +742,20 @@ export function createHostBoundary(options: HostBoundaryOptions): HostBoundary {
       // included (#2863). That waits behind `exited`, never inside it: a caller
       // that drains the streams only after the exit still can, and the wait is bounded.
       const settled = exited.then(() => worker.released(), () => worker.released())
-        .finally(() => { try { releaseOutput(); } finally { hold?.release(); } })
+        .finally(async () => { try { releaseOutput(); } finally { await hold?.release(); } })
         .catch(error => { try { options.log?.emit({ severityText: "WARN", body: "maintenance launch release refused", attributes: { error: String(error) } }); } catch { /* Logging cannot fail the settlement. */ } });
       const kill = (signal: NodeJS.Signals = "SIGTERM") => { signalCode = signal; worker.kill(signal); };
       execution.signal?.addEventListener("abort", () => kill("SIGKILL"), { once: true });
       // An abort accepted while the launch was awaited still stops the worker.
       if (execution.signal?.aborted) kill("SIGKILL");
       return retainExecutionAdmission({ stdout: worker.stdout, stderr: worker.stderr, exited, settled, get signalCode() { return signalCode; }, kill }, credential);
     },
-    runSync(execution) {
+    async runSync(execution) {
       if (execution.credentials) throw new Error("Credentialed maintenance requires asynchronous execution.");
       if (!isolated()) {
         if (execution.publish === false) throw Error("A discarded read requires the qualified worker boundary");
-        admitHost(null);
+        await admitHost(null);
+        stillHostFallback();
         const result = Bun.spawnSync([...execution.argv], { cwd: brain, env: { ...execution.env }, stdout: "pipe", stderr: "pipe", timeout: execution.timeoutMs });
         return { stdout: result.stdout.toString(), stderr: result.stderr.toString(), exitCode: result.exitCode ?? 1 };
       }

packages/ui-server/src/brain/host-execution.ts

@@ -17,6 +17,7 @@ import { createUiDb } from "../db/client.js";
 import { createHostBoundary, type HostBoundary } from "./host-boundary.js";
 import { assertReconciled, createGovernedManifests, type GovernedManifests, type Reconciliation } from "./governed-manifest.js";
 import { createWriterGate, type WriterGate } from "./writer-gate.js";
+import type { SqliteWriter } from "../db/write-worker.js";
 
 /**
  * Where quarantined files go: a host-private directory beside the UI
@@ -62,7 +63,7 @@ export interface HostExecution {
    */
   reconcileExisting(): Reconciliation[];
   /** Reconciles every abandoned governed turn on `db` (refusing if any path stayed), then opens the gate and manifests on it. */
-  attach(db: Database): Reconciliation[];
+  attach(db: Database, writer?: SqliteWriter): Reconciliation[];
   /** Closes the early handle `reconcileExisting` kept, when a boot refuses before `attach`. */
   release(): void;
 }
@@ -111,15 +112,16 @@ export function createHostExecution(options: { brainPath: string; dbPath: string
         return results;
       } catch (error) { db.close(); throw error; }
     },
-    attach(db) {
+    attach(db, writer) {
       // Reconciled first: a failure refuses before the gate exists, so no
       // gated maintenance can start on a brain still holding such a path.
       const opened = open(db);
       const results = opened.reconcileAbandoned();
       history = opened.hasHistory();
       assertReconciled(results);
       manifests = opened;
-      gate = createWriterGate(db, options.brainPath);
+      // Its transactions commit on the app's write worker (#2850).
+      gate = createWriterGate(db, options.brainPath, writer ? { writer } : {});
       early?.close();
       early = undefined;
       return results;