diff --git a/.changeset/writer-gate-off-event-loop.md b/.changeset/writer-gate-off-event-loop.md new file mode 100644 index 0000000000..d02be2f050 --- /dev/null +++ b/.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. diff --git a/docs/decisions/governed-writable-workers.md b/docs/decisions/governed-writable-workers.md index c2c5635912..de817b9293 100644 --- a/docs/decisions/governed-writable-workers.md +++ b/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 diff --git a/docs/decisions/sqlite-write-worker.md b/docs/decisions/sqlite-write-worker.md index 04c0d6d844..207bfb432a 100644 --- a/docs/decisions/sqlite-write-worker.md +++ b/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 diff --git a/docs/plans/storage-lifecycle.md b/docs/plans/storage-lifecycle.md index e15533a262..209953cc4f 100644 --- a/docs/plans/storage-lifecycle.md +++ b/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. diff --git a/packages/ui-server/src/app.ts b/packages/ui-server/src/app.ts index 07f56fff30..b798be84c8 100644 --- a/packages/ui-server/src/app.ts +++ b/packages/ui-server/src/app.ts @@ -453,7 +453,7 @@ export async function createApp(options: CreateAppOptions = {}): Promise 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 { 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()); diff --git a/packages/ui-server/src/brain/governed-turn.ts b/packages/ui-server/src/brain/governed-turn.ts index 5358d6dc0c..ff00ac95bf 100644 --- a/packages/ui-server/src/brain/governed-turn.ts +++ b/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((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]! }))), diff --git a/packages/ui-server/src/brain/host-boundary.ts b/packages/ui-server/src/brain/host-boundary.ts index bd82bd047a..50c6ea79f4 100644 --- a/packages/ui-server/src/brain/host-boundary.ts +++ b/packages/ui-server/src/brain/host-boundary.ts @@ -208,8 +208,11 @@ export interface HostBoundary { }): Promise; /** The same, with live output, for a streamed sync or a cron job. */ launch(execution: BrainExecution): Promise; - /** A short synchronous query (git `check-ignore`). Takes no gate. */ - runSync(execution: Omit & { 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 & { timeoutMs: number; startupView?: WorkerStartupView }): Promise; /** Runs `fn` while holding the gate, when there is one. */ withGate(mode: GateMode, purpose: string, fn: () => Promise, signal?: AbortSignal): Promise; } @@ -311,7 +314,7 @@ function credentialOutput(stream: ReadableStream, 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 } | 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(mode: GateMode, purpose: string, fn: (hold: GateHold | null) => Promise, signal?: AbortSignal): Promise { 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 = (execution: BrainExecution, fn: (hold: GateHold | null) => Promise) => 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,7 +742,7 @@ 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 }); @@ -737,11 +750,12 @@ export function createHostBoundary(options: HostBoundaryOptions): HostBoundary { 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 }; } diff --git a/packages/ui-server/src/brain/host-execution.ts b/packages/ui-server/src/brain/host-execution.ts index d852541360..da035e1237 100644 --- a/packages/ui-server/src/brain/host-execution.ts +++ b/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,7 +112,7 @@ 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); @@ -119,7 +120,8 @@ export function createHostExecution(options: { brainPath: string; dbPath: string 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; diff --git a/packages/ui-server/src/brain/turn-revert.ts b/packages/ui-server/src/brain/turn-revert.ts index 2086ab3f34..a15b472dc9 100644 --- a/packages/ui-server/src/brain/turn-revert.ts +++ b/packages/ui-server/src/brain/turn-revert.ts @@ -151,7 +151,7 @@ export function createTurnReverts(db: Database, options: { const reason = (error as {reason?:string}).reason === "journal_capacity" ? "journal_capacity" : (error as {reason?:string}).reason === "journal_storage_unavailable" ? "journal_unavailable" : error instanceof RevertUnavailable ? "authorization_changed" : "operation_unavailable"; try { if (observed) await journal.finish(row.journal_turn_id); else await journal.finishReserved(row.journal_turn_id); } catch { /* An interrupted journal remains partial, never undoable by assumption. */ } try { await journal.writeOperation([journalStatement("UPDATE turn_journal_revert_operations SET state=?,completed_at=?,error=? WHERE id=?", observed ? "unknown" : "failed", Date.now(), reason, row.id)]); } catch { /* Readback will retain the original pending identity. */ } - } finally { hold?.release(); if (selected.some(path => path.state === "restored" || path.state === "failed")) { try { options.refresh(); } catch { /* The ordinary index refresh may be retried; file effects remain recorded. */ } } } + } finally { await hold?.release(); if (selected.some(path => path.state === "restored" || path.state === "failed")) { try { options.refresh(); } catch { /* The ordinary index refresh may be retried; file effects remain recorded. */ } } } } let starting: Promise = Promise.resolve(); async function begin(principalId: string, sessionId: string, sourceTurnId: string, request: TurnRevertRequest): Promise { diff --git a/packages/ui-server/src/brain/writer-gate-writes.ts b/packages/ui-server/src/brain/writer-gate-writes.ts new file mode 100644 index 0000000000..600adbb572 --- /dev/null +++ b/packages/ui-server/src/brain/writer-gate-writes.ts @@ -0,0 +1,15 @@ +/** + * The writer gate's transactions off the event loop (#2850): the write + * worker imports this module, and the gate (`writer-gate.ts`) submits each + * transaction to it. Its policy reader lives here, with the transactions + * that read the policy. + */ +import type { Database } from "bun:sqlite"; + +import { gateProcedures } from "./writer-gate.js"; + +export const url = import.meta.url; + +export function setup(db: Database, options: { brain: string }) { + return gateProcedures(db, options.brain); +} diff --git a/packages/ui-server/src/brain/writer-gate.ts b/packages/ui-server/src/brain/writer-gate.ts index a423eb2e6f..b8d73cbcf1 100644 --- a/packages/ui-server/src/brain/writer-gate.ts +++ b/packages/ui-server/src/brain/writer-gate.ts @@ -27,6 +27,7 @@ import { join } from "node:path"; import { randomUUID } from "node:crypto"; import { createGovernancePolicyReader, providerFolderRoots, retainedFolderRoots, type TurnGovernancePolicy } from "./governance-policy.js"; import { ensureRetainedDirectories } from "./index-directory.js"; +import { procedureCaller, type SqliteWriter, type WireError, type WriteProcedures } from "../db/write-worker.js"; export type GateMode = "shared" | "exclusive"; @@ -59,31 +60,41 @@ export interface GateHold { /** Planned retained folders, including active peers. Not evidence that a worker mounted them. */ readonly folderPaths?: readonly string[]; /** Extend this held plan to the exact admitted policy; never removes active peer roots. */ - retainFolders?(paths: readonly string[]): readonly string[]; + retainFolders?(paths: readonly string[]): Promise; /** Refuse known missing/unaccounted peer plans. Success is not mounted-worker attestation. */ - assertPeerFolderPlans?(paths: readonly string[]): void; + assertPeerFolderPlans?(paths: readonly string[]): Promise; /** * `assertPeerFolderPlans`, but wait up to `timeoutMs` for peers admitted * before this hold retained `paths` to finish (#2832); every other refusal, * and a peer still active at the bound, refuses as before. */ waitForPeerFolderPlans?(paths: readonly string[], options?: { signal?: AbortSignal; timeoutMs?: number }): Promise; - release(): void; + /** + * Submits the release at once (#2850): every later gate write of this + * process sees it. Settles once it has committed; an owner whose + * completion promises the hold is gone awaits it. + */ + release(): Promise; } export interface WriterGate { /** Waits until `mode` is admitted, then holds it until `release()`. */ acquire(mode: GateMode, purpose: string, options?: { signal?: AbortSignal; turn?: boolean; queueTurns?: boolean }): Promise; - /** Current holders, stale ones removed. */ + /** + * Current holders, stale ones removed. A diagnostic, read and pruned on the + * gate's own connection; nothing on a turn's path calls it. + */ holders(): { mode: GateMode; purpose: string; pid: number }[]; /** Refuse an ungated host process beside retained or unknown folder plans, or under a folder policy. */ - assertUngatedHostFallback(): void; + assertUngatedHostFallback(): Promise; /** * Record the folders admission translated the policy's type selectors to, * under that policy's hash (#2836). Every later acquisition under the same * hash plans them, in any process. It replaces the brain's earlier record. */ - rememberTypeFolders(policyHash: string | null, folders: readonly string[]): void; + rememberTypeFolders(policyHash: string | null, folders: readonly string[]): Promise; + /** Settles once every gate write submitted before it has settled. */ + settled(): Promise; } /** How often a waiting acquisition looks again. */ @@ -131,11 +142,12 @@ export function refuseHostFallback(policy: TurnGovernancePolicy, roots: readonly // Only a concrete, still-owned hold can record negative host provenance. // This never attests mounts or upgrades a plan into qualified authority. -const hostWriterMarkers = new WeakMap void>(); -export function markUnpinnedHostWriter(hold: GateHold): void { +const hostWriterMarkers = new WeakMap Promise>(); +/** Settles once the downgrade has committed; nothing may spawn before then. */ +export function markUnpinnedHostWriter(hold: GateHold): Promise { const mark = hostWriterMarkers.get(hold); if (!mark) throw new Error("Host writer folder provenance cannot be recorded with this gate"); - mark(); + return mark(); } function folderPlan(raw: string): { roots: readonly string[]; unpinnedHost: boolean } { @@ -176,39 +188,134 @@ function mergeFolderPlans(rows: { folder_topology: string | null }[], additions: return Object.freeze([...roots].sort()); } -export function createWriterGate(db: Database, brainPath: string, options: { pollMs?: number } = {}): WriterGate { - let brain: string; - try { brain = realpathSync(brainPath); } catch { brain = brainPath; } - const pollMs = options.pollMs ?? GATE_POLL_MS; +/** + * The gate's transactions, run on the connection that commits them: the + * app's write worker when it has one (#2850), else the gate's own `db`. Each + * keeps its `BEGIN IMMEDIATE`, so it is serialized against every other + * process's gate writer as before; the worker runs them in submission order. + * One policy reader serves every observation the gate makes. + */ +export function gateProcedures(db: Database, brain: string): WriteProcedures { const policyReader = createGovernancePolicyReader(brain); const prune = () => { const rows = db.query("SELECT id, pid, pid_start FROM writer_gate WHERE brain = ?").all(brain) as { id: string; pid: number; pid_start: string }[]; for (const row of rows) if (!processAlive(row.pid, row.pid_start)) db.query("DELETE FROM writer_gate WHERE id = ?").run(row.id); }; - const tryAcquire = (mode: GateMode, purpose: string, turn: boolean, waiter: string | null): { id: string; folderPaths: readonly string[] } | null => db.transaction(() => { - prune(); - const held = db.query("SELECT mode, folder_topology FROM writer_gate WHERE brain = ?").all(brain) as { mode: GateMode; folder_topology: string | null }[]; - const waiters = db.query("SELECT id, pid, pid_start FROM writer_gate_turn_waiters WHERE brain = ? ORDER BY queued_at, rowid").all(brain) as { id: string; pid: number; pid_start: string }[]; - for (const row of waiters) if (!processAlive(row.pid, row.pid_start)) db.query("DELETE FROM writer_gate_turn_waiters WHERE id = ?").run(row.id); - const first = waiters.find(row => processAlive(row.pid, row.pid_start)); - const blocked = mode === "exclusive" ? held.length > 0 || (waiter !== null && first?.id !== waiter) : held.some(row => row.mode === "exclusive") || (turn && first !== undefined); - if (blocked) return null; - const id = randomUUID(); - if (waiter) db.query("DELETE FROM writer_gate_turn_waiters WHERE id = ?").run(waiter); - const policy = policyReader.load(); - const folderPaths = mergeFolderPlans(held, retainedFolderRoots(policy.policy, recordedTypeFolders(db, brain, policy.hash))); - db.query("INSERT INTO writer_gate (id, brain, mode, purpose, pid, pid_start, acquired_at, folder_topology) VALUES (?, ?, ?, ?, ?, ?, ?, ?)") - .run(id, brain, mode, purpose, SELF.pid, SELF.start, Date.now(), JSON.stringify(folderPaths)); - return { id, folderPaths }; - }).immediate(); + const owned = (id: string, message: string) => { + const own = db.query("SELECT id FROM writer_gate WHERE id = ? AND brain = ? AND pid = ? AND pid_start = ?").get(id, brain, SELF.pid, SELF.start); + if (!own) throw new Error(message); + }; + return { + tryAcquire: (mode: GateMode, purpose: string, turn: boolean, waiter: string | null, at: number): { id: string; folderPaths: readonly string[] } | null => db.transaction(() => { + prune(); + const held = db.query("SELECT mode, folder_topology FROM writer_gate WHERE brain = ?").all(brain) as { mode: GateMode; folder_topology: string | null }[]; + const waiters = db.query("SELECT id, pid, pid_start FROM writer_gate_turn_waiters WHERE brain = ? ORDER BY queued_at, rowid").all(brain) as { id: string; pid: number; pid_start: string }[]; + for (const row of waiters) if (!processAlive(row.pid, row.pid_start)) db.query("DELETE FROM writer_gate_turn_waiters WHERE id = ?").run(row.id); + const first = waiters.find(row => processAlive(row.pid, row.pid_start)); + const blocked = mode === "exclusive" ? held.length > 0 || (waiter !== null && first?.id !== waiter) : held.some(row => row.mode === "exclusive") || (turn && first !== undefined); + if (blocked) return null; + const id = randomUUID(); + if (waiter) db.query("DELETE FROM writer_gate_turn_waiters WHERE id = ?").run(waiter); + const policy = policyReader.load(); + const folderPaths = mergeFolderPlans(held, retainedFolderRoots(policy.policy, recordedTypeFolders(db, brain, policy.hash))); + db.query("INSERT INTO writer_gate (id, brain, mode, purpose, pid, pid_start, acquired_at, folder_topology) VALUES (?, ?, ?, ?, ?, ?, ?, ?)") + .run(id, brain, mode, purpose, SELF.pid, SELF.start, at, JSON.stringify(folderPaths)); + return { id, folderPaths }; + }).immediate(), + enqueue(waiter: string, at: number) { + db.query("INSERT INTO writer_gate_turn_waiters (id, brain, pid, pid_start, queued_at) VALUES (?, ?, ?, ?, ?)").run(waiter, brain, SELF.pid, SELF.start, at); + }, + dequeue(waiter: string) { db.query("DELETE FROM writer_gate_turn_waiters WHERE id = ?").run(waiter); }, + // Throws for a peer that is unsafe whatever it does next; returns + // whether some peer's plan lacks a requested root. + peers: (id: string, requested: readonly string[]): boolean => db.transaction(() => { + prune(); + owned(id, "Writer folder plan no longer has its admitted owner"); + const peers = db.query("SELECT folder_topology FROM writer_gate WHERE brain = ? AND id <> ?").all(brain, id) as { folder_topology: string | null }[]; + let unretained = false; + for (const peer of peers) { + if (peer.folder_topology === null) throw new Error("An active peer has an unaccounted provider folder plan; finish that peer and retry admission"); + let plan: ReturnType; + try { plan = folderPlan(peer.folder_topology); } + catch { throw new Error("An active peer has an unverified provider folder plan; finish that peer and retry admission"); } + if (plan.unpinnedHost) throw new Error("An active peer is an unpinned host writer; finish that peer and retry admission"); + // A descendant root also pins each ancestor. An ancestor root + // alone does not retain an arbitrary child inside its subtree. + if (requested.some(path => !plan.roots.some(root => root === path || root.startsWith(`${path}/`)))) unretained = true; + } + return unretained; + }).immediate(), + retain: (id: string, folderPaths: readonly string[], paths: readonly string[]): readonly string[] => db.transaction(() => { + prune(); + owned(id, "Writer folder plan no longer has its admitted owner"); + const rows = db.query("SELECT folder_topology FROM writer_gate WHERE brain = ?").all(brain) as { folder_topology: string | null }[]; + const next = mergeFolderPlans(rows, [...folderPaths, ...paths]); + db.query("UPDATE writer_gate SET folder_topology = ? WHERE id = ?").run(JSON.stringify(next), id); + return next; + }).immediate(), + unpinned: (id: string, folderPaths: readonly string[]) => db.transaction(() => { + prune(); + owned(id, "Host writer folder plan no longer has its admitted owner"); + const peers = db.query("SELECT folder_topology FROM writer_gate WHERE brain = ? AND id <> ?").all(brain, id) as { folder_topology: string | null }[]; + for (const peer of peers) { + if (peer.folder_topology === null) throw new Error("Host fallback cannot enter beside an active peer with unknown folder provenance"); + const plan = folderPlan(peer.folder_topology); + if (!plan.unpinnedHost && (folderPaths.length || plan.roots.length)) + throw new Error("Host fallback cannot enter beside an active peer that may depend on retained provider folders"); + } + // Ruling (c) on #2284: no host process starts while folders may + // be pinned. The current policy is reread inside this transaction. + refuseHostFallback(policyReader.load(), mergeFolderPlans(peers, folderPaths)); + // Serialize this downgrade against governed admission, before + // the host process exists. The recorded roots are always empty. + db.query("UPDATE writer_gate SET folder_topology = ? WHERE id = ?").run(JSON.stringify({ kind: "unpinned_host", roots: folderPaths }), id); + }).immediate(), + release(id: string) { db.query("DELETE FROM writer_gate WHERE id = ?").run(id); }, + ungated: () => db.transaction(() => { + prune(); + const rows = db.query("SELECT folder_topology FROM writer_gate WHERE brain = ?").all(brain) as { folder_topology: string | null }[]; + if (rows.some(row => row.folder_topology === null)) + throw new HostFallbackRefusedError("Host fallback is refused beside an active writer with unknown folder provenance; finish that writer and retry."); + let roots: readonly string[]; + try { roots = mergeFolderPlans(rows, []); } + catch { throw new HostFallbackRefusedError("Host fallback is refused beside an active writer with an unverified folder plan; finish that writer and retry."); } + refuseHostFallback(policyReader.load(), roots); + }).immediate(), + rememberTypeFolders(policyHash: string, roots: readonly string[], at: number) { + db.query("INSERT INTO provider_type_folders (brain, policy_hash, folders, translated_at) VALUES (?, ?, ?, ?) ON CONFLICT (brain) DO UPDATE SET policy_hash = excluded.policy_hash, folders = excluded.folders, translated_at = excluded.translated_at") + .run(brain, policyHash, JSON.stringify(roots), at); + }, + }; +} + +const WRITES = new URL(`./writer-gate-writes${import.meta.url.endsWith(".ts") ? ".ts" : ".js"}`, import.meta.url).href; +/** A refusal keeps its class across the worker boundary. */ +const revive = (error: WireError) => error.name === "HostFallbackRefusedError" ? new HostFallbackRefusedError(error.message) : undefined; + +export function createWriterGate(db: Database, brainPath: string, options: { pollMs?: number; + /** Where the gate's transactions commit: the app's write worker (#2850). Absent: on `db`, at once. */ + writer?: SqliteWriter } = {}): WriterGate { + let brain: string; + try { brain = realpathSync(brainPath); } catch { brain = brainPath; } + const pollMs = options.pollMs ?? GATE_POLL_MS; + const call = procedureCaller(db, { url: WRITES, setup: (connection: Database, setupOptions: { brain: string }) => gateProcedures(connection, setupOptions.brain) }, + { brain }, options.writer, revive); + // Every gate write submitted and not yet settled: `settled()` waits for them. + const pending = new Set>(); + const submit = (name: string, ...args: unknown[]): Promise => { + const done = call(name, ...args); + pending.add(done); + void done.then(() => {}, () => {}).then(() => pending.delete(done)); + return done; + }; return { async acquire(mode, purpose, { signal, turn = false, queueTurns = false } = {}) { signal?.throwIfAborted(); const waiter = mode === "exclusive" && queueTurns ? randomUUID() : null; - if (waiter) db.query("INSERT INTO writer_gate_turn_waiters (id, brain, pid, pid_start, queued_at) VALUES (?, ?, ?, ?, ?)").run(waiter, brain, SELF.pid, SELF.start, Date.now()); + if (waiter) await submit("enqueue", waiter, Date.now()); try { for (;;) { signal?.throwIfAborted(); - const acquired = tryAcquire(mode, purpose, turn, waiter); + const acquired = await submit<{ id: string; folderPaths: readonly string[] } | null>("tryAcquire", mode, purpose, turn, waiter, Date.now()); if (acquired) { const { id } = acquired; // A plan retaining the index or cache directory needs it to exist; @@ -216,48 +323,22 @@ export function createWriterGate(db: Database, brainPath: string, options: { pol try { ensureRetainedDirectories(brain, acquired.folderPaths); } catch { /* resolver refuses */ } let folderPaths = acquired.folderPaths; let released = false, unpinnedHost = false; - // Throws for a peer that is unsafe whatever it does next; returns - // whether some peer's plan lacks a requested root. - const checkPeerFolderPlans = (paths: readonly string[]): boolean => { + let releasing: Promise | undefined; + const checkPeerFolderPlans = async (paths: readonly string[]): Promise => { if (released) throw new Error("Writer folder plan was already released"); const requested = mergeFolderPlans([], paths); if (!requested.length) return false; if (unpinnedHost) throw new Error("An unpinned host writer cannot qualify provider folder admission"); if (requested.some(root => !folderPaths.includes(root))) throw new Error("Requested writer folder plan is not held"); - return db.transaction(() => { - prune(); - const own = db.query("SELECT id FROM writer_gate WHERE id = ? AND brain = ? AND pid = ? AND pid_start = ?").get(id, brain, SELF.pid, SELF.start); - if (!own) throw new Error("Writer folder plan no longer has its admitted owner"); - const peers = db.query("SELECT folder_topology FROM writer_gate WHERE brain = ? AND id <> ?").all(brain, id) as { folder_topology: string | null }[]; - let unretained = false; - for (const peer of peers) { - if (peer.folder_topology === null) throw new Error("An active peer has an unaccounted provider folder plan; finish that peer and retry admission"); - let plan: ReturnType; - try { plan = folderPlan(peer.folder_topology); } - catch { throw new Error("An active peer has an unverified provider folder plan; finish that peer and retry admission"); } - if (plan.unpinnedHost) throw new Error("An active peer is an unpinned host writer; finish that peer and retry admission"); - // A descendant root also pins each ancestor. An ancestor root - // alone does not retain an arbitrary child inside its subtree. - if (requested.some(path => !plan.roots.some(root => root === path || root.startsWith(`${path}/`)))) unretained = true; - } - return unretained; - }).immediate(); + return submit("peers", id, requested); }; - const hold: GateHold = { id, mode, get folderPaths() { return folderPaths; }, retainFolders(paths) { + const hold: GateHold = { id, mode, get folderPaths() { return folderPaths; }, async retainFolders(paths) { if (released) throw new Error("Writer folder plan was already released"); if (unpinnedHost) throw new Error("An unpinned host writer cannot upgrade its folder plan"); - folderPaths = db.transaction(() => { - prune(); - const own = db.query("SELECT id FROM writer_gate WHERE id = ? AND brain = ? AND pid = ? AND pid_start = ?").get(id, brain, SELF.pid, SELF.start); - if (!own) throw new Error("Writer folder plan no longer has its admitted owner"); - const rows = db.query("SELECT folder_topology FROM writer_gate WHERE brain = ?").all(brain) as { folder_topology: string | null }[]; - const next = mergeFolderPlans(rows, [...folderPaths, ...paths]); - db.query("UPDATE writer_gate SET folder_topology = ? WHERE id = ?").run(JSON.stringify(next), id); - return next; - }).immediate(); + folderPaths = await submit("retain", id, folderPaths, paths); return folderPaths; - }, assertPeerFolderPlans(paths) { - if (checkPeerFolderPlans(paths)) throw new Error(UNRETAINED_PEER); + }, async assertPeerFolderPlans(paths) { + if (await checkPeerFolderPlans(paths)) throw new Error(UNRETAINED_PEER); }, async waitForPeerFolderPlans(paths, { signal, timeoutMs = PEER_FOLDER_PLAN_WAIT_MS } = {}) { // Every writer admitted after this hold retained `paths` merges // them into its own plan, so only earlier ones can lack them, and @@ -265,69 +346,45 @@ export function createWriterGate(db: Database, brainPath: string, options: { pol const deadline = Date.now() + timeoutMs; for (;;) { signal?.throwIfAborted(); - if (!checkPeerFolderPlans(paths)) return; + if (!await checkPeerFolderPlans(paths)) return; if (Date.now() >= deadline) throw new Error(UNRETAINED_PEER); await pause(Math.min(pollMs, Math.max(0, deadline - Date.now())), signal); } }, release() { - if (released) return; + // Submitted at once, so every later gate write of this process + // sees it; the returned promise settles once it has committed. + if (releasing) return releasing; released = true; hostWriterMarkers.delete(hold); - db.query("DELETE FROM writer_gate WHERE id = ?").run(id); + releasing = submit("release", id); + // A caller that does not wait still never leaves it unhandled. + releasing.catch(() => {}); + return releasing; } }; - hostWriterMarkers.set(hold, () => { + hostWriterMarkers.set(hold, async () => { if (released) throw new Error("Writer folder plan was already released"); if (unpinnedHost) return; - db.transaction(() => { - prune(); - const own = db.query("SELECT id FROM writer_gate WHERE id = ? AND brain = ? AND pid = ? AND pid_start = ?").get(id, brain, SELF.pid, SELF.start); - if (!own) throw new Error("Host writer folder plan no longer has its admitted owner"); - const peers = db.query("SELECT folder_topology FROM writer_gate WHERE brain = ? AND id <> ?").all(brain, id) as { folder_topology: string | null }[]; - for (const peer of peers) { - if (peer.folder_topology === null) throw new Error("Host fallback cannot enter beside an active peer with unknown folder provenance"); - const plan = folderPlan(peer.folder_topology); - if (!plan.unpinnedHost && (folderPaths.length || plan.roots.length)) - throw new Error("Host fallback cannot enter beside an active peer that may depend on retained provider folders"); - } - // Ruling (c) on #2284: no host process starts while folders may - // be pinned. The current policy is reread inside this transaction. - refuseHostFallback(policyReader.load(), mergeFolderPlans(peers, folderPaths)); - // Serialize this downgrade against governed admission, before - // the host process exists. The recorded roots are always empty. - db.query("UPDATE writer_gate SET folder_topology = ? WHERE id = ?").run(JSON.stringify({ kind: "unpinned_host", roots: folderPaths }), id); - }).immediate(); + await submit("unpinned", id, folderPaths); unpinnedHost = true; }); return hold; } await pause(pollMs, signal); } } finally { - if (waiter) db.query("DELETE FROM writer_gate_turn_waiters WHERE id = ?").run(waiter); + if (waiter) await submit("dequeue", waiter); } }, - assertUngatedHostFallback() { - db.transaction(() => { - prune(); - const rows = db.query("SELECT folder_topology FROM writer_gate WHERE brain = ?").all(brain) as { folder_topology: string | null }[]; - if (rows.some(row => row.folder_topology === null)) - throw new HostFallbackRefusedError("Host fallback is refused beside an active writer with unknown folder provenance; finish that writer and retry."); - let roots: readonly string[]; - try { roots = mergeFolderPlans(rows, []); } - catch { throw new HostFallbackRefusedError("Host fallback is refused beside an active writer with an unverified folder plan; finish that writer and retry."); } - refuseHostFallback(policyReader.load(), roots); - }).immediate(); - }, - rememberTypeFolders(policyHash, folders) { + assertUngatedHostFallback: () => submit("ungated"), + async rememberTypeFolders(policyHash, folders) { if (policyHash === null) return; const roots = folderPlan(JSON.stringify([...new Set(folders)].sort())).roots; - db.query("INSERT INTO provider_type_folders (brain, policy_hash, folders, translated_at) VALUES (?, ?, ?, ?) ON CONFLICT (brain) DO UPDATE SET policy_hash = excluded.policy_hash, folders = excluded.folders, translated_at = excluded.translated_at") - .run(brain, policyHash, JSON.stringify(roots), Date.now()); - }, - holders() { - return db.transaction(() => { - prune(); - return db.query("SELECT mode, purpose, pid FROM writer_gate WHERE brain = ? ORDER BY acquired_at").all(brain) as { mode: GateMode; purpose: string; pid: number }[]; - }).immediate(); + await submit("rememberTypeFolders", policyHash, roots, Date.now()); }, + holders: () => db.transaction(() => { + const rows = db.query("SELECT id, pid, pid_start FROM writer_gate WHERE brain = ?").all(brain) as { id: string; pid: number; pid_start: string }[]; + for (const row of rows) if (!processAlive(row.pid, row.pid_start)) db.query("DELETE FROM writer_gate WHERE id = ?").run(row.id); + return db.query("SELECT mode, purpose, pid FROM writer_gate WHERE brain = ? ORDER BY acquired_at").all(brain) as { mode: GateMode; purpose: string; pid: number }[]; + }).immediate(), + async settled() { while (pending.size) await Promise.allSettled([...pending]); }, }; } diff --git a/packages/ui-server/src/storage/removal.ts b/packages/ui-server/src/storage/removal.ts index c6434a0096..234aac2406 100644 --- a/packages/ui-server/src/storage/removal.ts +++ b/packages/ui-server/src/storage/removal.ts @@ -104,7 +104,7 @@ export function createStorageRemoval(db: Database, options: { async function exclusive(signal: AbortSignal, purpose: string, apply: () => Promise): Promise { const hold = await options.gate.acquire("exclusive", purpose, { queueTurns: true, signal }); try { return await options.lock.withLock(BRAIN_LOCK_KEY, () => options.lock.withLock(GIT_LOCK_KEY, apply, { signal }), { signal }); } - finally { hold.release(); } + finally { await hold.release(); } } function pathsFor(paths: string[]): string[] { if (!Array.isArray(paths) || paths.length < 1 || paths.length > STORAGE_REMOVAL_LIMITS.items || paths.some(path => typeof path !== "string" || Buffer.byteLength(path) > 4096 || /[\\\u0000-\u001f\u007f]/u.test(path) || path.split("/").some(part => !part || part === "." || part === "..")) || new Set(paths).size !== paths.length) throw new StorageUnavailable(400, "Select one bounded exact set of attachment paths."); diff --git a/packages/ui-server/tests/credential-maintenance-exclusion.test.ts b/packages/ui-server/tests/credential-maintenance-exclusion.test.ts index f7750104dd..2cd4e6c487 100644 --- a/packages/ui-server/tests/credential-maintenance-exclusion.test.ts +++ b/packages/ui-server/tests/credential-maintenance-exclusion.test.ts @@ -37,7 +37,7 @@ test("real isolated credentialed maintenance admits its writer before waiting, e let entered!: () => void; const admitted = new Promise(resolve => { entered = resolve; }); let writerHeld = false; - const gate = { async acquire() { writerHeld = true; entered(); return { id: "odysseus", mode: "exclusive" as const, release() { writerHeld = false; } }; }, holders: () => [], assertUngatedHostFallback() {}, rememberTypeFolders() {} }; + const gate = { async acquire() { writerHeld = true; entered(); return { id: "odysseus", mode: "exclusive" as const, async release() { writerHeld = false; } }; }, holders: () => [], async assertUngatedHostFallback() {}, async rememberTypeFolders() {}, async settled() {} }; const boundary = createHostBoundary({ brainPath: f.brain, nativeManager: createNativeBrainManagerClient(native.nativeBrain), dbPath: join(f.root, "state/ui.db"), home: f.home, credentialDomain: f.domain, gate: () => gate }); const held = await acquireHostCredentialExclusion({ domain: f.domain }); cleanup.push(() => held.release()); @@ -78,7 +78,7 @@ test("cancelled credential admission releases writer admission without releasing const native = createNativeAppFixture(f.brain); cleanup.push(native.close); let entered!: () => void; const admitted = new Promise(resolve => { entered = resolve; }); let writerHeld = false; - const gate = { async acquire() { writerHeld = true; entered(); return { id: "odysseus", mode: "exclusive" as const, release() { writerHeld = false; } }; }, holders: () => [], assertUngatedHostFallback() {}, rememberTypeFolders() {} }; + const gate = { async acquire() { writerHeld = true; entered(); return { id: "odysseus", mode: "exclusive" as const, async release() { writerHeld = false; } }; }, holders: () => [], async assertUngatedHostFallback() {}, async rememberTypeFolders() {}, async settled() {} }; const boundary = createHostBoundary({ brainPath: f.brain, dbPath: join(f.root, "state/ui.db"), nativeManager: createNativeBrainManagerClient(native.nativeBrain), credentialDomain: f.domain, gate: () => gate }); const hold = await acquireHostCredentialExclusion({ domain: f.domain }); cleanup.push(() => hold.release()); const abort = new AbortController(); diff --git a/packages/ui-server/tests/host-fallback-gates.test.ts b/packages/ui-server/tests/host-fallback-gates.test.ts index 3c2fd34595..0ef20b2a15 100644 --- a/packages/ui-server/tests/host-fallback-gates.test.ts +++ b/packages/ui-server/tests/host-fallback-gates.test.ts @@ -308,7 +308,7 @@ test("a synchronous host reader refuses under a provider folder policy and runs try { const query = { argv: [process.execPath, "--config=/dev/null", "--no-env-file", "-e", `require("node:fs").writeFileSync(${JSON.stringify(f.ready)},"read");console.log("Odysseus reads")`], env: { PATH: "/usr/bin:/bin" }, credentials: false, purpose: "Odysseus query", timeoutMs: 5000 }; - expect(f.boundary().runSync(query), "NO_FOLDER_POLICY_KEEPS_SYNC_HOST_READER").toEqual({ stdout: "Odysseus reads\n", stderr: "", exitCode: 0 }); + expect(await f.boundary().runSync(query), "NO_FOLDER_POLICY_KEEPS_SYNC_HOST_READER").toEqual({ stdout: "Odysseus reads\n", stderr: "", exitCode: 0 }); rmSync(f.ready); mkdirSync(join(f.brain, "context/policies"), { recursive: true }); writeFileSync(join(f.brain, "context/policies/governance.md"), "---\nschema_version: 1\nproviders: [{ paths: [group/private/**], allow: [anthropic] }]\n---\nAthena restricts a folder.\n"); diff --git a/packages/ui-server/tests/maintenance-probe-admission.test.ts b/packages/ui-server/tests/maintenance-probe-admission.test.ts index a025a033dd..effcc8c6b5 100644 --- a/packages/ui-server/tests/maintenance-probe-admission.test.ts +++ b/packages/ui-server/tests/maintenance-probe-admission.test.ts @@ -32,6 +32,6 @@ test("an uncompleted maintenance probe reports busy without throwing or selectin } finally { held.release(); } await boundary.prepare(); expect(boundary.status(), "BUSY_PROBE_RECOVERS_WITHOUT_A_MINUTE_CACHE").toMatchObject({ available: true, hostExecution: "worker" }); - const result = boundary.runSync({ argv: ["/bin/sh", "-c", "printf READMITTED"], env: {}, credentials: false, purpose: "admission after release", timeoutMs: 5000, startupView: maintenance.startupView }); + const result = await boundary.runSync({ argv: ["/bin/sh", "-c", "printf READMITTED"], env: {}, credentials: false, purpose: "admission after release", timeoutMs: 5000, startupView: maintenance.startupView }); expect(result.exitCode, result.stderr).toBe(0); expect(result.stdout).toBe("READMITTED"); }); diff --git a/packages/ui-server/tests/native-host-inputs.test.ts b/packages/ui-server/tests/native-host-inputs.test.ts index d625e15cb4..2ba192215d 100644 --- a/packages/ui-server/tests/native-host-inputs.test.ts +++ b/packages/ui-server/tests/native-host-inputs.test.ts @@ -29,7 +29,7 @@ test("sync readers require an explicitly prepared owner and never borrow a boot expect(() => boundary.runSync({ ...f.execution, timeoutMs: 1000 }), "NO_SYNC_PREPARATION_OR_BOOT_REUSE").toThrow("caller's prepared immutable startupView"); const view = await prepareNativeWorkerStartupView(f.brain, f.manager); fs.writeFileSync(join(f.brain, "helper.txt"), "Penelope later live input"); - const result = boundary.runSync({ ...f.execution, startupView: view, timeoutMs: 1000 }); + const result = await boundary.runSync({ ...f.execution, startupView: view, timeoutMs: 1000 }); expect(result.exitCode, result.stderr).toBe(0); expect(result.stdout.trim()).toBe("Odysseus initial input"); expect(f.released.length).toBe(0); await view.cleanup(); expect(f.released).toEqual(f.checkpoints); } finally { f.cleanup(); } diff --git a/packages/ui-server/tests/private-index.test.ts b/packages/ui-server/tests/private-index.test.ts index 688d8532a6..a9c4dd2446 100644 --- a/packages/ui-server/tests/private-index.test.ts +++ b/packages/ui-server/tests/private-index.test.ts @@ -6,6 +6,7 @@ */ import { installFixtureNativeStartupViews } from "./helpers/fixture-native-startup.js"; installFixtureNativeStartupViews(); +import { expectRejection } from "./helpers/expect-rejection.js"; import { afterEach, expect, spyOn, test, type Mock } from "bun:test"; import { cpSync, existsSync, linkSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, renameSync, rmSync, symlinkSync, writeFileSync } from "node:fs"; import { join, resolve } from "node:path"; @@ -54,7 +55,7 @@ function fixture(providers = "[{ paths: [me/**], allow: [anthropic] }]", profile const argv = typeof command === "function" ? command(brain) : command; const turns = createGovernedTurns({ nativeManager: createNativeJournalFixture(brain, root).client, brainPath: brain, execution, profile, maskPaths: [], index: async () => {}, report: message => reports.push(message), ...(argv ? { privateIndexCommand: argv } : {}) }); - return { root, brain, turns, reports, execution }; + return { root, brain, db, turns, reports, execution }; } function search(brain: string, sandbox: Parameters[0]["sandbox"], query: string): string[] { @@ -217,6 +218,15 @@ test("a writer in another process, such as the cron runner, plans the type folde } }, 90_000); +test("a type selector's translation that cannot be recorded refuses the turn (#2850)", async () => { + const f = fixture("[{ types: [identity], allow: [anthropic] }]"); + // Later writers, in any process, plan only what is recorded: an unrecorded + // translation is not admitted beside them. + f.db.exec("DROP TABLE provider_type_folders"); + await expectRejection(f.turns.begin("unrecorded", new AbortController().signal), /provider_type_folders/, "TYPE_FOLDER_RECORD_FAILURE_REFUSES_ADMISSION"); + expect(f.execution.gate()!.holders(), "TYPE_FOLDER_REFUSAL_RELEASES_ITS_HOLD").toEqual([]); +}); + test("a folder nested inside another masked folder launches under the outer mask alone", async () => { // Bubblewrap cannot create me/sub inside the empty read-only tmpfs over me, // so a second mask for it would fail every launch (#2718). diff --git a/packages/ui-server/tests/writer-gate-offloop.test.ts b/packages/ui-server/tests/writer-gate-offloop.test.ts new file mode 100644 index 0000000000..e43884fc5f --- /dev/null +++ b/packages/ui-server/tests/writer-gate-offloop.test.ts @@ -0,0 +1,128 @@ +/** The writer gate's transactions commit on the write worker (#2850). */ +import { expect, test } from "bun:test"; +import { existsSync, mkdirSync, mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createHostBoundary } from "../src/brain/host-boundary.js"; +import { createWriterGate, HostFallbackRefusedError, markUnpinnedHostWriter } from "../src/brain/writer-gate.js"; +import { createUiDb } from "../src/db/client.js"; +import { openSqliteWriter } from "../src/db/write-worker.js"; +import * as probe from "./fixtures/write-worker-probe.js"; +import { expectRejection } from "./helpers/expect-rejection.js"; +import { watchWriteSites } from "./helpers/write-sites.js"; + +function fixture() { + const root = mkdtempSync(join(tmpdir(), "writer-gate-odysseus-")), brain = join(root, "brain"), path = join(root, "ui.db"); + mkdirSync(brain); + const db = createUiDb(path), writer = openSqliteWriter(db); + return { root, brain, path, db, writer, async close() { await writer.close(); db.close(); rmSync(root, { recursive: true, force: true }); } }; +} +const GATE_TABLES = ["writer_gate", "writer_gate_turn_waiters", "provider_type_folders"]; +const row = (f: ReturnType, id: string) => f.db.query("SELECT folder_topology FROM writer_gate WHERE id = ?").get(id) as { folder_topology: string } | null; +/** A worker commit held for `ms`, as a slow fsync would; call `hold` to queue one ahead of what follows. */ +const slowCommits = async (f: ReturnType) => { + const channel = f.writer.channel(probe, { table: "odysseus_log" }); await channel.ready; + return { hold: (ms: number) => channel.call("slow", "Calypso", ms) }; +}; + +test("the gate's transactions commit on the write worker at FULL, never on the app connection", async () => { + const f = fixture(); + try { + expect(f.writer.offloop).toBe(true); + const sites = await watchWriteSites(f.db, f.writer, GATE_TABLES); + const gate = createWriterGate(f.db, f.brain, { writer: f.writer, pollMs: 5 }); + const turn = await gate.acquire("shared", "turn:voyage", { turn: true }); + await gate.rememberTypeFolders("policy-hash", ["notes"]); + expect(await turn.retainFolders!([]), "GATE_RETAIN_COMMITS_ON_WORKER").toEqual([]); + await turn.assertPeerFolderPlans!([]); + await turn.release(); + const revert = await gate.acquire("exclusive", "revert:ithaca", { queueTurns: true }); + await revert.release(); + const host = await gate.acquire("shared", "host:odysseus"); + await markUnpinnedHostWriter(host); + expect(JSON.parse(row(f, host.id)!.folder_topology), "GATE_UNPINNED_COMMITS_ON_WORKER").toEqual({ kind: "unpinned_host", roots: [] }); + await host.release(); + await gate.assertUngatedHostFallback(); + await gate.settled(); + const log = await sites.log(); + expect(new Set(log), "GATE_TABLES_WRITTEN_ON_WORKER").toEqual(new Set(GATE_TABLES)); + expect(gate.holders(), "GATE_RELEASES_COMMITTED").toEqual([]); + } finally { await f.close(); } +}); + +test("a release counts for this process's next acquisition at once, and settles once it has committed", async () => { + const f = fixture(); + try { + // A poll this long would time the test out: the next acquisition must not need one. + const gate = createWriterGate(f.db, f.brain, { writer: f.writer, pollMs: 60_000 }); + const first = await gate.acquire("exclusive", "storage:odysseus"); + const commits = await slowCommits(f); + const slow = commits.hold(300); + const released = first.release(); + expect(row(f, first.id), "GATE_RELEASE_NOT_COMMITTED_ON_LOOP").not.toBeNull(); + let settled = false; + void released.then(() => { settled = true; }); + const next = gate.acquire("exclusive", "revert:ithaca"); + await Bun.sleep(50); + expect(settled, "GATE_RELEASE_SETTLES_AFTER_ITS_COMMIT").toBe(false); + await slow; + await released; + expect(row(f, first.id), "GATE_RELEASE_COMMITTED").toBeNull(); + // Without seeing the release it would wait a whole poll. + expect(await Promise.race([next.then(() => "acquired"), Bun.sleep(5_000).then(() => "polled")]), "GATE_NEXT_ACQUISITION_SEES_EARLIER_RELEASE").toBe("acquired"); + await (await next).release(); + } finally { await f.close(); } +}); + +test("a host writer runs only after its downgrade commits, and not if the boundary changed meanwhile", async () => { + const f = fixture(); + try { + const gate = createWriterGate(f.db, f.brain, { writer: f.writer, pollMs: 5 }); + // Downgrade order: the promise settles only after the row says unpinned. + const hold = await gate.acquire("shared", "host:odysseus"); + const commits = await slowCommits(f); + let slow = commits.hold(300); + const marked = markUnpinnedHostWriter(hold); + expect(JSON.parse(row(f, hold.id)!.folder_topology), "GATE_UNPINNED_NOT_YET_COMMITTED").toEqual([]); + await marked; + expect(JSON.parse(row(f, hold.id)!.folder_topology), "GATE_UNPINNED_COMMITTED_BEFORE_SETTLING").toEqual({ kind: "unpinned_host", roots: [] }); + await slow; + await hold.release(); + // The boundary turns isolated while the downgrade waits behind a slow + // commit, queued just before it: nothing spawns, on either host path. + for (const method of ["run", "launch"] as const) { + let history = false; + const ran = join(f.root, `ran-${method}`); + // Once the hold is granted, a slow commit queues ahead of the downgrade + // and the boundary turns isolated while it waits. + const racing = { ...gate, async acquire(...args: Parameters) { + const granted = await gate.acquire(...args); + slow = commits.hold(300); setTimeout(() => { history = true; }, 50); + return granted; + } }; + const boundary = createHostBoundary({ brainPath: f.brain, dbPath: f.path, requireGate: true, gate: () => racing, history: () => history, + workerHost: () => ({ ok: false, requirement: "fictional never-governed fixture" }) }); + expect(boundary.isolated()).toBe(false); + const execution = { argv: [process.execPath, "--config=/dev/null", "--no-env-file", "-e", `require("node:fs").writeFileSync(${JSON.stringify(ran)}, "ran")`], + env: { PATH: "/usr/bin:/bin" }, credentials: false, gate: "shared" as const, purpose: "Odysseus host writer" }; + await expectRejection(boundary[method](execution), "worker boundary changed", `GATE_HOST_REFUSED_WHEN_BOUNDARY_CHANGED_DURING_ADMISSION_${method}`); + expect(existsSync(ran), `GATE_HOST_CHILD_NEVER_SPAWNED_${method}`).toBe(false); + await slow; + await gate.settled(); + expect(gate.holders(), `GATE_REFUSED_HOST_RELEASED_${method}`).toEqual([]); + } + } finally { await f.close(); } +}); + +test("a policy refusal keeps its class across the worker", async () => { + const f = fixture(); + try { + mkdirSync(join(f.brain, "context/policies"), { recursive: true }); + await Bun.write(join(f.brain, "context/policies/governance.md"), "---\nschema_version: 1\nproviders: [\n---\n"); + const gate = createWriterGate(f.db, f.brain, { writer: f.writer, pollMs: 5 }); + const host = await gate.acquire("shared", "host:odysseus"); + await expectRejection(markUnpinnedHostWriter(host), HostFallbackRefusedError, "GATE_REFUSAL_CLASS_KEPT_ACROSS_WORKER"); + await expectRejection(gate.assertUngatedHostFallback(), HostFallbackRefusedError, "GATE_UNGATED_REFUSAL_CLASS_KEPT_ACROSS_WORKER"); + await host.release(); + } finally { await f.close(); } +}); diff --git a/packages/ui-server/tests/writer-gate.test.ts b/packages/ui-server/tests/writer-gate.test.ts index 4f8c80caab..9dc566a345 100644 --- a/packages/ui-server/tests/writer-gate.test.ts +++ b/packages/ui-server/tests/writer-gate.test.ts @@ -238,7 +238,7 @@ describe("provider folder admission beside earlier writers (#2832)", () => { const w = world(); const peer = await w.cron.acquire("shared", "brain CLI version"); expect(peer.folderPaths, "EARLIER_WRITER_PLANS_NO_FOLDER").toEqual([]); - if (host) markUnpinnedHostWriter(peer); + if (host) await markUnpinnedHostWriter(peer); mkdirSync(join(w.brain, "context/policies"), { recursive: true }); mkdirSync(join(w.brain, "group/private"), { recursive: true }); writeFileSync(join(w.brain, "context/policies/governance.md"), "---\nschema_version: 1\nproviders: [{ paths: [group/private/**], allow: [anthropic] }]\n---\nAthena restricts a folder.\n"); @@ -310,9 +310,9 @@ describe("unpinned host writer provenance", () => { const w = world(); try { const first = await w.server.acquire("shared", "host:odysseus"); - markUnpinnedHostWriter(first); + await markUnpinnedHostWriter(first); const second = await w.cron.acquire("shared", "host:athena"); - markUnpinnedHostWriter(second); + await markUnpinnedHostWriter(second); expect(w.server.holders()).toHaveLength(2); // A folder policy written while the hosts run applies to later admissions. writePolicy(w, "providers: [{ paths: [group/private/**], allow: [anthropic] }]\n"); @@ -330,7 +330,7 @@ describe("unpinned host writer provenance", () => { try { const host = await w.server.acquire("shared", "host:odysseus"); let refusal: unknown; - try { markUnpinnedHostWriter(host); } catch (error) { refusal = error; } + try { await markUnpinnedHostWriter(host); } catch (error) { refusal = error; } expect(refusal, "FOLDER_POLICY_REFUSES_HOST_CLASSIFICATION").toBeInstanceOf(HostFallbackRefusedError); expect(String(refusal)).toContain("group/private"); expect(w.db.query("SELECT folder_topology FROM writer_gate WHERE id = ?").get(host.id), "REFUSED_HOST_KEEPS_ITS_PLAN") @@ -420,7 +420,7 @@ describe("unpinned host writer provenance", () => { try { const host = await w.server.acquire("shared", "host:odysseus"); expect(() => markUnpinnedHostWriter({ ...host })).toThrow("cannot be recorded"); - markUnpinnedHostWriter(host); + await markUnpinnedHostWriter(host); expect(() => host.retainFolders!(["group/private"])).toThrow("cannot upgrade"); expect(() => host.assertPeerFolderPlans!(["group/private"])).toThrow("cannot qualify"); expect(w.db.query("SELECT folder_topology FROM writer_gate WHERE id = ?").get(host.id)).toEqual({ folder_topology: JSON.stringify({ kind: "unpinned_host", roots: [] }) }); diff --git a/packages/ui-server/tests/ws-reference-worktrees.test.ts b/packages/ui-server/tests/ws-reference-worktrees.test.ts index 8148993fcb..b7a9bcd981 100644 --- a/packages/ui-server/tests/ws-reference-worktrees.test.ts +++ b/packages/ui-server/tests/ws-reference-worktrees.test.ts @@ -100,7 +100,8 @@ test("a turn's tools reach reference source and its own worktree through the bou await handlers.onOpen(undefined as never, ws); handlers.onMessage({ data: JSON.stringify({ type: "chat_message", text: "Read the Ithaca charts", requestId: "reference-request" }) } as MessageEvent, ws); const until = Date.now() + 10_000; - while (calls.length < 2 || app.wsHost.referenceWorktreeGenerations.size) { if (Date.now() > until) throw new Error(JSON.stringify({ calls, sent })); await Bun.sleep(5); } + // The turn ends once its gate release has committed (#2850), after its generation retires. + while (calls.length < 2 || app.wsHost.referenceWorktreeGenerations.size || app.wsHost.coordinator.isTurnActive()) { if (Date.now() > until) throw new Error(JSON.stringify({ calls, sent })); await Bun.sleep(5); } expect(sandbox?.references, "REFERENCE_TOOLS_SANDBOX_BINDS_VIEWS").toEqual(sources.views(principal.id) as { public: string; private: string }); const [read, requested] = calls as [Extract, Extract]; expect(read.source.path, "REFERENCE_TOOLS_HOST_TURN_SOURCE_PATH").toMatch(/^\/tmp\/brain-reference\/public\//);