fix(ui-server): commit local exchanges, classified blocks and saved worker preferences on the write worker

.changeset/app-writes-chat-leftovers-off-event-loop.md

@@ -0,0 +1,5 @@
+---
+"@schlessera/brain-ui-server": patch
+---
+
+Changed: the hosted server commits local exchanges, a turn's classified blocks and classifier confidence, and a saved worker preference (`PUT …/profile`) on its SQLite write worker at `synchronous=FULL` instead of on the event loop (#2850). A local exchange is still carried by the next prompt while it commits, and is answered once stored; a replayed history waits for the classified blocks submitted before it.

docs/decisions/answer-delivery.md

@@ -82,7 +82,7 @@ carries, no longer by position.
 
 The host also re-sends a session's pending questions after `session_resume`
 (`resendPendingAsks(host, ws, msg.sessionId)`,
-`packages/ui-server/src/ws/dispatch.ts:648`). Before this, a resume replaced
+`packages/ui-server/src/ws/dispatch.ts:649`). Before this, a resume replaced
 the transcript and left only a history card behind, which was the second
 symptom.
 

docs/decisions/governed-writable-workers.md

@@ -150,7 +150,7 @@ interactive work. Voice shows its policy read-only; dictation is interactive.
 The current worker never changes profile because a later save succeeded.
 Unknown enforcement facts remain unknown, and requested preference is separate
 from the actual admitted worker, including fallback
-(`createSessionProfiles`, `packages/ui-server/src/ws/session-profiles.ts:16-132`;
+(`createSessionProfiles`, `packages/ui-server/src/ws/session-profiles.ts:16-119`;
 `begin`, `packages/ui-server/src/brain/governed-turn.ts:145-311`).
 
 **Where a turn falls back to `strict`.** The launcher refuses to start

docs/decisions/http-api-boundary.md

@@ -46,11 +46,11 @@ explicit supported rows, not silently excluded by this distinction.
 
 The source audit on 2026-09-30 used
 `e0875269b21b707482545be9e94901acb897893b`. The public health mount precedes auth
-(`createHealthRoutes`, `packages/ui-server/src/app.ts:869`), the API guard follows
-public ceremonies (`authGuard`, `packages/ui-server/src/app.ts:890`), and the
-WebSocket admission is separate (`"/ws"`, `packages/ui-server/src/app.ts:1080`).
+(`createHealthRoutes`, `packages/ui-server/src/app.ts:870`), the API guard follows
+public ceremonies (`authGuard`, `packages/ui-server/src/app.ts:891`), and the
+WebSocket admission is separate (`"/ws"`, `packages/ui-server/src/app.ts:1081`).
 Static serving is conditional (`if (options.staticRoot)`,
-`packages/ui-server/src/app.ts:1100-1121`). That order establishes access;
+`packages/ui-server/src/app.ts:1101-1122`). That order establishes access;
 it does not select compatibility status.
 
 A hermetic real-app launch, with temporary brain root, in-memory database,

docs/decisions/session-principals.md

@@ -96,9 +96,9 @@ Both reviewers verified every row below against the source.
 | Cookie verify | `auth.ts:265-296` | strict `^(\d+)\.(\d+)$`, safe integers, future/expired rejection, `cookieEpoch === sessionsEpoch(db)` |
 | Global epoch | `auth.ts:302-318` | `settings` row; missing = 0; non-integer **throws** |
 | Global revoke | `auth.ts:321-330` | `setSetting` (overflow-guarded) + `clients.closeAll(1008, …)` |
-| Auth routes | `authRoutes`, `app.ts:875` | mounted **before** the guard at `authGuard(authMode`, `app.ts:890`; logout guards itself (`resolveCookiePrincipal(c, auth, deps.db)`, `auth.ts:799`) |
-| Passkey management | `passkeyManagementRoutes`, `app.ts:900` | mounted **after** the guard ("Mount AFTER the auth guard") |
-| WS guard | `"/ws"`, `app.ts:1080-1097` | origin → `isWsAuthorized` (returns a boolean, `auth.ts:221-237`) → capacity → upgrade |
+| Auth routes | `authRoutes`, `app.ts:876` | mounted **before** the guard at `authGuard(authMode`, `app.ts:891`; logout guards itself (`resolveCookiePrincipal(c, auth, deps.db)`, `auth.ts:799`) |
+| Passkey management | `passkeyManagementRoutes`, `app.ts:901` | mounted **after** the guard ("Mount AFTER the auth guard") |
+| WS guard | `"/ws"`, `app.ts:1081-1098` | origin → `isWsAuthorized` (returns a boolean, `auth.ts:221-237`) → capacity → upgrade |
 | WS upgrade | `ws/connection.ts:262-264` | `createWsUpgrade` ignores the request context |
 | WS admission | `ws/connection.ts:84` | `clients.add(ws)` — the socket's identity is unknown |
 | Turn record | `ws/run-session.ts:136-144` | recorder built from `{turnId, sessionId, billing}`; `RunningTurn` (`ws/turns.ts:55-77`) has no actor |

docs/decisions/sqlite-write-worker.md

@@ -158,8 +158,8 @@ are in [the governed-workers decision](governed-writable-workers.md#admitted-obj
   acceptance's revision is assigned where it is accepted, one above the
   highest the catalog has issued or seen committed, so it is known at once
   and a failed write's revision is never issued again; only the newest
-  acceptance's outcome orders or unorders the session. Retained retries and
-  local exchanges still commit on the app connection, so a failed retry
+  acceptance's outcome orders or unorders the session. Retained retries
+  still commit on the app connection, so a failed retry
   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.
@@ -206,14 +206,21 @@ are in [the governed-workers decision](governed-writable-workers.md#admitted-obj
   A read of a session's exposures (admission's continuation checks, a
   handoff's history check) first waits for every exposure write submitted
   before it. Boot reconciliation still closes abandoned manifests inline.
+- **Local exchanges, classified blocks and saved worker preferences
+  followed** (#2850). A local exchange commits on the worker with the
+  catalog's other writes. Until it has, the next prompt handed over still
+  takes it, exactly once, and the hand-over is recorded on the worker after
+  the exchange's own record; the client is answered once it is stored. A
+  turn's classified blocks, the blocks that follow a promoted link and the
+  classifier's confidence commit on the worker
+  (`packages/ui-server/src/classification/classification-writes.ts`), and a
+  replayed history waits for every one submitted before it. A saved worker
+  preference (`PUT …/profile`) is checked and stored in one immediate
+  transaction on the worker, and its refusals keep their class and status.
 - **Other app-connection writers stay on the loop for now**: retained
-  retries and local exchanges, saved worker profiles, policy
-  proposals, action and storage notices, push subscriptions and VAPID keys,
-  principal revocation and administration, passkeys, inbox and settings. Each commits at
-- **Other app-connection writers stay on the loop for now**: retained
-  retries and local exchanges, saved worker profiles, policy
-  proposals, action and storage notices, push subscriptions and VAPID keys,
-  principal revocation and administration, passkeys, inbox and settings. Each commits at
+  retries, drafts, policy proposals, action and storage notices, push
+  subscriptions and VAPID keys, principal revocation and administration,
+  passkeys, inbox and settings. Each commits at
   FULL on the loop, and while a worker procedure holds its turn, such a write
   also waits there for the worker's commit to finish, including a journal
   procedure that writes only `journal.db`, because the ledger's

docs/decisions/voice-permission.md

@@ -396,7 +396,7 @@ exchange over a security decision is a second chance for noise to produce a
 grant.
 
 **"Always allow" cannot be given by voice.** It is a persistent policy change
-(the block computing `remembers`, `packages/ui-server/src/ws/dispatch.ts:446-452`)
+(the block computing `remembers`, `packages/ui-server/src/ws/dispatch.ts:447-453`)
 and it is the one decision on
 the card with no keyboard shortcut, by D37's ruling 5, and the reason given
 there is exactly the one that applies here: *"a letter that grants standing
@@ -405,7 +405,7 @@ permission by reflex is the one footgun in the vocabulary"*
 A microphone is a reflex surface with worse recognition than a keyboard. This
 costs nothing measurable:
 the server already refuses `always` for kind `command` requests
-(the block computing `remembers`, `dispatch.ts:446-452`, and the lookup computing
+(the block computing `remembers`, `dispatch.ts:447-453`, and the lookup computing
 `remembered`, `ws/bridge.ts:368-388`), and 192 of 192 measured
 approvals were kind `command`.
 
@@ -475,7 +475,7 @@ provenance.
 
 The wire needs nothing new: a spoken refusal is an ordinary `tool_denial`
 (in `handleClientMessage`, the arm `case "tool_denial"`,
-`packages/ui-server/src/ws/dispatch.ts:493-516`) with a message naming the
+`packages/ui-server/src/ws/dispatch.ts:494-517`) with a message naming the
 phrase that produced it.
 
 ## Containment: shared with #51, deliberately not identical
@@ -610,7 +610,7 @@ matters because a reader cannot otherwise tell a live hazard from a fixed one.
   left out (the lookup computing `remembered`,
   `packages/ui-server/src/ws/bridge.ts:368-388`,
   and the block computing `remembers`,
-  `packages/ui-server/src/ws/dispatch.ts:446-452`).
+  `packages/ui-server/src/ws/dispatch.ts:447-453`).
   The evaluation happens
   before the lookup, which is what this record asked for.
 

docs/investigations/live-conversation-investigation.md

@@ -279,12 +279,12 @@ matcher. An isolated ambiguous refusal denies all **current** pending requests;
 an exact refusal can name one. With none pending, it is ordinary input. Bind
 each denial to existing IDs, echoed turn and `channel:"voice"`; the server still
 rejects voice grants
-(`if (msg.channel === "voice")`, `packages/ui-server/src/ws/dispatch.ts:406-429`).
+(`if (msg.channel === "voice")`, `packages/ui-server/src/ws/dispatch.ts:407-430`).
 Never add a spoken approve or always-allow path.
 
 On the evidence date the denial handler removed/resolved a matched request but
 provided no dedicated denial receipt at that branch; #957 added one
-(`case "tool_denial"`, `packages/ui-server/src/ws/dispatch.ts:493-516`). The
+(`case "tool_denial"`, `packages/ui-server/src/ws/dispatch.ts:494-517`). The
 integration therefore needs host-authoritative resolution/replay: denied,
 already granted, expired, or unknown/disconnected. Speak a denial confirmation
 only for a confirmed denied outcome. A visual grant winning the race remains

docs/plans/async-collaboration-requirements.md

@@ -226,7 +226,7 @@ Stated before the requirements because five of them derive from it.
   that is the shape to copy, including its `close()` lifecycle.
 - R19. **The cron backstop has independent authorization before the general guard.**
   Mount it on the existing listener before
-  (`app.use("/api/*", authGuard(`, `packages/ui-server/src/app.ts:890`).
+  (`app.use("/api/*", authGuard(`, `packages/ui-server/src/app.ts:891`).
   Authorize a boot-minted ephemeral token, rotated each boot and stored in a
   0600 runtime file, with the actual socket address as an additional check.
   Proxy headers cannot authorize it. The poke succeeds in every auth mode

docs/plans/async-collaboration.md

@@ -131,7 +131,7 @@ system gate. It is not a prerequisite to rebuild or a new independent package.
   owns manual triggers/history; it does not tick the inbox.
 - **Two connections:** (`export function createUiDb(`, `packages/ui-server/src/db/client.ts:27-44`) sets WAL,
   foreign keys and a 5-second busy timeout. Claims are immediate transactions.
-- **Auth mounting:** (`app.use("/api/*", authGuard(`, `packages/ui-server/src/app.ts:890`) follows public routes;
+- **Auth mounting:** (`app.use("/api/*", authGuard(`, `packages/ui-server/src/app.ts:891`) follows public routes;
   (`export function authGuard(`, `packages/ui-server/src/middleware/auth.ts:185-245`) binds principals in each auth mode.
   An internal poke needs independent token authorization before this guard.
 - **Permission parking:** (`requestPermission: (req) => {`, `packages/ui-server/src/ws/bridge.ts:358-452`). Timeout unwind is

docs/plans/navigation-recovery.md

@@ -385,7 +385,7 @@ implementation. Any PR moving these lines owns their citation repairs.
 
 - Host reconnect: (`const runningTurns = [...coordinator.running].filter((t) => t.sessionId);`, `packages/ui-server/src/ws/connection.ts:126-191`).
 
-- Host resume: (`case "session_resume": {`, `packages/ui-server/src/ws/dispatch.ts:628-682`).
+- Host resume: (`case "session_resume": {`, `packages/ui-server/src/ws/dispatch.ts:629-683`).
 
 - Catalog activity: (`persistSessionStub(sessionId, promptText, providerId, backendId) {`, `packages/ui-server/src/ws/session-catalog.ts:369-376`).
 
@@ -397,7 +397,7 @@ implementation. Any PR moving these lines owns their citation repairs.
 
 - Activity resolution: (`.get("/activity/runs", (c) => {`, `packages/ui-server/src/routes/activity.ts:143-281`).
 
-- API authorization: (`app.use("/api/*", authGuard(authMode, auth, db));`, `packages/ui-server/src/app.ts:890`).
+- API authorization: (`app.use("/api/*", authGuard(authMode, auth, db));`, `packages/ui-server/src/app.ts:891`).
 
 - Queued acceptance: (`// Queue it as the session's next turn; report queued immediately.`, `packages/ui-server/src/ws/run-session.ts:1098-1143`).
 

packages/ui-server/src/app.ts

@@ -601,6 +601,7 @@ export async function createApp(options: CreateAppOptions = {}): Promise<BrainUi
     db: () => db,
     log: observability.logger("classification"),
     meter: observability.meter("classification"),
+    writer,
   });
   // Per-session composer drafts (#979): the UI's operational database, one
   // namespace for this host, shared by its principals.

packages/ui-server/src/classification/blocks-store.ts

@@ -26,13 +26,14 @@ export function saveMessageBlocks(
   db: Database,
   sessionId: string,
   partText: string,
-  blocks: MessageBlock[]
+  blocks: MessageBlock[],
+  at = Date.now()
 ): void {
   db.prepare(
     `INSERT INTO message_blocks (session_id, part_hash, blocks, created_at)
      VALUES (?, ?, ?, ?)
      ON CONFLICT(session_id, part_hash) DO UPDATE SET blocks = excluded.blocks, created_at = excluded.created_at`
-  ).run(sessionId, partHash(partText), JSON.stringify(blocks), Date.now());
+  ).run(sessionId, partHash(partText), JSON.stringify(blocks), at);
 }
 
 export function loadMessageBlocks(db: Database, sessionId: string, partText: string): MessageBlock[] {

packages/ui-server/src/classification/classification-writes.ts

@@ -0,0 +1,24 @@
+/**
+ * Classified blocks and classifier confidence off the event loop (#2850): a
+ * turn's classified blocks, the blocks that follow a promoted link and the
+ * confidence behind a pass commit on the `ui.db` write worker at
+ * `synchronous=FULL`, through the writers that ran inline. Replay joins
+ * blocks only after every write submitted before it has settled.
+ */
+import type { Database } from "bun:sqlite";
+import type { MessageBlock } from "@schlessera/brain-ui-sdk/protocol";
+
+import type { WriteProcedures } from "../db/write-worker.js";
+import { rewriteMessageBlockLinks, saveMessageBlocks } from "./blocks-store.js";
+import type { QuestionObservation } from "@schlessera/brain-ui-sdk/internal";
+import { recordQuestionConfidence } from "./confidence-store.js";
+
+export const url = import.meta.url;
+
+export function setup(db: Database): WriteProcedures {
+  return {
+    saveBlocks: (sessionId: string, partText: string, blocks: MessageBlock[], at: number) => saveMessageBlocks(db, sessionId, partText, blocks, at),
+    rewriteLinks: (sessionId: string, textParts: string[], links: [string, string][]) => rewriteMessageBlockLinks(db, sessionId, textParts, new Map(links)),
+    confidence: (sessionId: string, observations: QuestionObservation[], at: number) => recordQuestionConfidence(db, sessionId, observations, at),
+  };
+}

packages/ui-server/src/classification/classify-turn.ts

@@ -29,8 +29,9 @@ import {
   type ClassificationPlan,
 } from "@schlessera/brain-ui-sdk/internal";
 
-import { attachMessageBlocks, rewriteMessageBlockLinks, saveMessageBlocks } from "./blocks-store.js";
-import { recordQuestionConfidence } from "./confidence-store.js";
+import { attachMessageBlocks } from "./blocks-store.js";
+import * as classificationWrites from "./classification-writes.js";
+import { procedureCaller, type ProcedureCall, type SqliteWriter } from "../db/write-worker.js";
 import type { JevClient, JevOutcome } from "./jev-client.js";
 
 type PartKind = "text" | "thinking" | "tool";
@@ -74,6 +75,8 @@ export interface ClassificationPassDeps {
   db: () => Database;
   log: Logger;
   meter?: Meter;
+  /** Where blocks and confidence commit: the app's write worker (#2850). Absent: on `db`, at once. */
+  writer?: SqliteWriter;
 }
 
 export interface ClassificationPassRecord {
@@ -99,9 +102,20 @@ export interface TurnClassifier {
    * removed.
    */
   attach(sessionId: string, messages: SessionHistoryMessage[]): SessionHistoryMessage[];
+  /** Settles once every write submitted before it has settled: replay joins blocks after it (#2850). */
+  settled?(): Promise<void>;
 }
 
 export function createTurnClassifier(deps: ClassificationPassDeps): TurnClassifier {
+  let caller: ProcedureCall | undefined;
+  let tail: Promise<void> = Promise.resolve();
+  const submit = (name: string, ...args: unknown[]): Promise<void> => {
+    let done: Promise<void>;
+    try { caller ??= procedureCaller(deps.db(), classificationWrites, undefined, deps.writer); done = caller<void>(name, ...args); }
+    catch (error) { done = Promise.reject(error); }
+    tail = done.then(() => {}, () => {});
+    return done;
+  };
   const passes = deps.meter?.createCounter("classification.passes", {
     description: "Classification passes by outcome",
   });
@@ -121,8 +135,12 @@ export function createTurnClassifier(deps: ClassificationPassDeps): TurnClassifi
     blocks: MessageBlock[]
   ): void {
     try {
-      recordQuestionConfidence(deps.db(), sessionId, observeClassification(plan, answers, blocks));
+      submit("confidence", sessionId, observeClassification(plan, answers, blocks), Date.now()).catch(confidenceFailed(sessionId));
     } catch (err) {
+      confidenceFailed(sessionId)(err);
+    }
+  }
+  const confidenceFailed = (sessionId: string) => (err: unknown) => {
       deps.log.emit({
         severityText: "WARN",
         body: "classifier confidence could not be recorded",
@@ -131,8 +149,7 @@ export function createTurnClassifier(deps: ClassificationPassDeps): TurnClassifi
           "error.message": err instanceof Error ? err.message : String(err),
         },
       });
-    }
-  }
+  };
 
   function record(sessionId: string, entry: ClassificationPassRecord): void {
     passes?.add(1, { outcome: entry.outcome });
@@ -159,16 +176,16 @@ export function createTurnClassifier(deps: ClassificationPassDeps): TurnClassifi
   return {
     enabled: deps.jev.enabled,
     rewriteLinks(sessionId, textParts, links) {
-      try {
-        rewriteMessageBlockLinks(deps.db(), sessionId, textParts, links);
-      } catch (err) {
+      if (links.size === 0) return;
+      submit("rewriteLinks", sessionId, [...textParts], [...links]).catch((err: unknown) => {
         deps.log.emit({
           severityText: "WARN",
           body: "classified blocks could not follow promoted links",
           attributes: { "session.id": sessionId, "error.message": err instanceof Error ? err.message : String(err) },
         });
-      }
+      });
     },
+    async settled() { let waiting: Promise<void>; do { waiting = tail; await waiting; } while (waiting !== tail); },
     attach(sessionId, messages) {
       try {
         return attachMessageBlocks(deps.db(), sessionId, messages);
@@ -220,7 +237,8 @@ export function createTurnClassifier(deps: ClassificationPassDeps): TurnClassifi
         }
         for (const [partIndex, list] of byPart) {
           const text = textParts[partIndex];
-          if (text !== undefined) saveMessageBlocks(deps.db(), sessionId, text, list);
+          // Committed off the event loop (#2850); replay waits for it.
+          if (text !== undefined) await submit("saveBlocks", sessionId, text, list, Date.now());
         }
         return blocks;
       } catch (err) {

packages/ui-server/src/ws/dispatch.ts

@@ -282,7 +282,8 @@ export async function handleClientMessage(
       // A command the client answered itself (#582). Stored against the
       // session; the next prompt handed to the backend carries its context.
       // Answered either way, so the client can say when it is not kept.
-      const saved = catalog.recordLocalExchange?.(msg.sessionId, msg.exchange, false) ?? false;
+      // Answered once it is stored (#2850); the next prompt takes it meanwhile.
+      const saved = (await catalog.recordLocalExchange?.(msg.sessionId, msg.exchange, false)) ?? false;
       host.sendMessage(ws, {
         type: "local_exchange_result",
         sessionId: msg.sessionId,