diff --git a/.changeset/app-writes-chat-leftovers-off-event-loop.md b/.changeset/app-writes-chat-leftovers-off-event-loop.md new file mode 100644 index 0000000000..4d02444a69 --- /dev/null +++ b/.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. diff --git a/docs/decisions/answer-delivery.md b/docs/decisions/answer-delivery.md index 6b8abd14d7..c670cdbe76 100644 --- a/docs/decisions/answer-delivery.md +++ b/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. diff --git a/docs/decisions/governed-writable-workers.md b/docs/decisions/governed-writable-workers.md index 03dd8d21a9..4be4996e5a 100644 --- a/docs/decisions/governed-writable-workers.md +++ b/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 diff --git a/docs/decisions/http-api-boundary.md b/docs/decisions/http-api-boundary.md index 1265573329..cf38e0cf59 100644 --- a/docs/decisions/http-api-boundary.md +++ b/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, diff --git a/docs/decisions/session-principals.md b/docs/decisions/session-principals.md index 0edc1fde11..2cdfc1fd52 100644 --- a/docs/decisions/session-principals.md +++ b/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 | diff --git a/docs/decisions/sqlite-write-worker.md b/docs/decisions/sqlite-write-worker.md index 01b32eb1b6..7aec23effc 100644 --- a/docs/decisions/sqlite-write-worker.md +++ b/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 diff --git a/docs/decisions/voice-permission.md b/docs/decisions/voice-permission.md index 99bedf6705..edc34f5223 100644 --- a/docs/decisions/voice-permission.md +++ b/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. diff --git a/docs/investigations/live-conversation-investigation.md b/docs/investigations/live-conversation-investigation.md index 6fc51740f2..c1194eee49 100644 --- a/docs/investigations/live-conversation-investigation.md +++ b/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 diff --git a/docs/plans/async-collaboration-requirements.md b/docs/plans/async-collaboration-requirements.md index eb5cf71671..feb7bb8526 100644 --- a/docs/plans/async-collaboration-requirements.md +++ b/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 diff --git a/docs/plans/async-collaboration.md b/docs/plans/async-collaboration.md index 95b3fd57af..4f35251b13 100644 --- a/docs/plans/async-collaboration.md +++ b/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 diff --git a/docs/plans/navigation-recovery.md b/docs/plans/navigation-recovery.md index bb600d950b..9eba7889f4 100644 --- a/docs/plans/navigation-recovery.md +++ b/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`). diff --git a/packages/ui-server/src/app.ts b/packages/ui-server/src/app.ts index 40d1c462bd..458ff7e727 100644 --- a/packages/ui-server/src/app.ts +++ b/packages/ui-server/src/app.ts @@ -601,6 +601,7 @@ export async function createApp(options: CreateAppOptions = {}): Promise 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. diff --git a/packages/ui-server/src/classification/blocks-store.ts b/packages/ui-server/src/classification/blocks-store.ts index 73f3fd945d..4d16474f09 100644 --- a/packages/ui-server/src/classification/blocks-store.ts +++ b/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[] { diff --git a/packages/ui-server/src/classification/classification-writes.ts b/packages/ui-server/src/classification/classification-writes.ts new file mode 100644 index 0000000000..a66e668bf3 --- /dev/null +++ b/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), + }; +} diff --git a/packages/ui-server/src/classification/classify-turn.ts b/packages/ui-server/src/classification/classify-turn.ts index 17477e345b..364c997fb6 100644 --- a/packages/ui-server/src/classification/classify-turn.ts +++ b/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; } export function createTurnClassifier(deps: ClassificationPassDeps): TurnClassifier { + let caller: ProcedureCall | undefined; + let tail: Promise = Promise.resolve(); + const submit = (name: string, ...args: unknown[]): Promise => { + let done: Promise; + try { caller ??= procedureCaller(deps.db(), classificationWrites, undefined, deps.writer); done = caller(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; 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) { diff --git a/packages/ui-server/src/ws/dispatch.ts b/packages/ui-server/src/ws/dispatch.ts index b140d213f0..aeb513fd67 100644 --- a/packages/ui-server/src/ws/dispatch.ts +++ b/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, diff --git a/packages/ui-server/src/ws/host.ts b/packages/ui-server/src/ws/host.ts index 5c28da6fd1..f9e52823c7 100644 --- a/packages/ui-server/src/ws/host.ts +++ b/packages/ui-server/src/ws/host.ts @@ -600,7 +600,7 @@ export class WsHost { * @internal Host wiring. */ async historySettled(): Promise { - await Promise.all([this.catalog.settled?.(), this.policyProposals?.settled()]); + await Promise.all([this.catalog.settled?.(), this.policyProposals?.settled(), this.classifier?.settled?.()]); } /** diff --git a/packages/ui-server/src/ws/local-exchanges.ts b/packages/ui-server/src/ws/local-exchanges.ts index 57aa5f8d6b..762e37fa79 100644 --- a/packages/ui-server/src/ws/local-exchanges.ts +++ b/packages/ui-server/src/ws/local-exchanges.ts @@ -42,9 +42,9 @@ export function saveLocalExchange( db: Database, sessionId: string, exchange: LocalExchange, - delivered: boolean + delivered: boolean, + now = Date.now() ): void { - const now = Date.now(); db.prepare( `INSERT INTO local_exchanges (session_id, exchange_id, command, prompt, answer, context, created_at, delivered_at) @@ -86,6 +86,19 @@ export function takePendingLocalExchanges(db: Database, sessionId: string): Loca }).immediate(); } +/** + * Mark exactly these exchanges carried (#2850): the catalog takes pending + * exchanges where the prompt is handed over and records the hand-over here, + * on the write worker, after the records it names. + */ +export function markLocalExchangesDelivered(db: Database, sessionId: string, exchangeIds: readonly string[], at: number): void { + if (!exchangeIds.length) return; + db.prepare( + `UPDATE local_exchanges SET delivered_at = ? WHERE session_id = ? AND delivered_at IS NULL + AND exchange_id IN (${exchangeIds.map(() => "?").join(", ")})` + ).run(at, sessionId, ...exchangeIds); +} + /** Every exchange recorded for a session, in order. */ export function loadLocalExchanges(db: Database, sessionId: string): LocalExchangeRecord[] { const rows = db diff --git a/packages/ui-server/src/ws/run-session.ts b/packages/ui-server/src/ws/run-session.ts index 8b5681c8f9..0766e09535 100644 --- a/packages/ui-server/src/ws/run-session.ts +++ b/packages/ui-server/src/ws/run-session.ts @@ -160,14 +160,14 @@ function leaseFor(authorization: AuthorizationContext, work: HostWork | undefine */ function recordDraftExchanges(host: WsHost, sessionId: string, exchanges: readonly LocalExchange[]): void { for (const exchange of exchanges) { - const saved = host.catalog.recordLocalExchange?.(sessionId, exchange, true) ?? false; - host.sendToClients({ + // Each is answered once it is stored (#2850). + void (host.catalog.recordLocalExchange?.(sessionId, exchange, true) ?? Promise.resolve(false)).then(saved => host.sendToClients({ type: "local_exchange_result", sessionId, exchangeId: exchange.id, saved, ...(saved ? {} : { reason: "The server could not store it." }), - }); + })); } } @@ -1198,14 +1198,14 @@ export async function handleChatMessage( // carries them, whichever of the paths below hands it over. if (sessionId && localExchanges?.length) { for (const exchange of localExchanges) { - const saved = host.catalog.recordLocalExchange?.(sessionId, exchange, false) ?? false; - host.sendMessage(ws, { + // Recorded now, so the prompt handed over next takes it; answered once stored (#2850). + void (host.catalog.recordLocalExchange?.(sessionId, exchange, false) ?? Promise.resolve(false)).then(saved => host.sendMessage(ws, { type: "local_exchange_result", sessionId, exchangeId: exchange.id, saved, ...(saved ? {} : { reason: "The server could not store it." }), - }); + })); } } const runningTurn = sessionId ? coordinator.bySession.get(sessionId) : undefined; diff --git a/packages/ui-server/src/ws/session-catalog-writes.ts b/packages/ui-server/src/ws/session-catalog-writes.ts index 45884aadce..9bc4037773 100644 --- a/packages/ui-server/src/ws/session-catalog-writes.ts +++ b/packages/ui-server/src/ws/session-catalog-writes.ts @@ -2,17 +2,17 @@ * Session catalog writes off the event loop (#2850). The hosted server's app * connection no longer commits the session row, accepted work, turn * boundaries, message sources, turn failures, the state room, Claude usage - * accounting, handoff links, labels or session costs itself: they commit on + * accounting, handoff links, labels, local exchanges or session costs itself: they commit on * the `ui.db` write worker at `synchronous=FULL`, in submission order with the * journal's, Activity's, governance, scope and budget writes. Each procedure * runs the statement or transaction that ran inline before. The catalog * (`session-catalog.ts`) answers its own synchronous reads from what it has * submitted until each write settles, and `settled()` holds the readers that - * join history onto the committed rows. Retained retries and local exchanges - * still commit on the app connection. + * join history onto the committed rows. Retained retries still commit on the + * app connection. */ import type { Database } from "bun:sqlite"; -import type { BillingMode, MessageSource, PricingRoute, SharedFileMeta, StateRoomObservation, ThinkingLevel } from "@schlessera/brain-ui-sdk/protocol"; +import type { BillingMode, LocalExchange, MessageSource, PricingRoute, SharedFileMeta, StateRoomObservation, ThinkingLevel } from "@schlessera/brain-ui-sdk/protocol"; import type { WriteProcedures } from "../db/write-worker.js"; import { createActivityStore, priceUsage, type ActivityStore, type RollupPricing } from "../activity/store.js"; @@ -21,6 +21,7 @@ import { createModelPricing } from "../pricing/model-pricing.js"; import { resolveStandalonePricingConfig } from "../config/env.js"; import { allocateClaudeTitleUsage, reconcileClaudeUsage, type ClaudeUsageEvidence, type ClaudeUsageState } from "./claude-usage-accounting.js"; import { saveMessageSource, saveEffectiveThinkingLevel } from "./message-sources.js"; +import { markLocalExchangesDelivered, saveLocalExchange } from "./local-exchanges.js"; import * as work from "./session-work.js"; import { saveTurnFailure, type RecordedTurnFailure } from "./turn-failures.js"; @@ -158,6 +159,8 @@ export function setup(db: Database, options: SessionCatalogWriteOptions = {}): W label(sessionId: string, label: string, source: string) { db.prepare("UPDATE sessions SET label = ?, label_source = ? WHERE id = ?").run(label, source, sessionId); }, + localExchange: (sessionId: string, exchange: LocalExchange, delivered: boolean, at: number) => saveLocalExchange(db, sessionId, exchange, delivered, at), + exchangesDelivered: (sessionId: string, exchangeIds: string[], at: number) => markLocalExchangesDelivered(db, sessionId, exchangeIds, at), addCost(sessionId: string, costUsd: number) { db.prepare("UPDATE sessions SET total_cost_usd = COALESCE(total_cost_usd, 0) + ? WHERE id = ?").run(costUsd, sessionId); }, diff --git a/packages/ui-server/src/ws/session-catalog.ts b/packages/ui-server/src/ws/session-catalog.ts index d24b80bde4..c9bd805ebd 100644 --- a/packages/ui-server/src/ws/session-catalog.ts +++ b/packages/ui-server/src/ws/session-catalog.ts @@ -22,8 +22,6 @@ import { stateRoomObservationSchema } from "@schlessera/brain-ui-sdk/schemas"; import { attachMessageSources } from "./message-sources.js"; import { loadLocalExchanges, - saveLocalExchange, - takePendingLocalExchanges, type LocalExchangeRecord, } from "./local-exchanges.js"; @@ -121,7 +119,7 @@ export interface SessionCatalog { * handed over now. Returns whether it is stored. Optional like the source * seam: without it the exchange stays on the client that ran it. */ - recordLocalExchange?(sessionId: string, exchange: LocalExchange, delivered: boolean): boolean; + recordLocalExchange?(sessionId: string, exchange: LocalExchange, delivered: boolean): Promise; /** The exchanges no prompt has carried yet, marked as carried by the one being handed over. */ takePendingLocalExchanges?(sessionId: string): LocalExchange[]; /** Every exchange recorded for the session, for replay. */ @@ -213,6 +211,8 @@ export function createSessionCatalog(db: () => Database, log?: Logger, scratchPi // own write settles: a failed one stops counting at once. type RowEffect = { upsert: { providerId: string | null; backendId: string | null } } | { label: { label: string; source: string } }; const rowEffects = new Map(); + const pendingExchanges = new Map(); + const takenExchanges = new Map>(); const pendingWork = new Map work.AcceptedWork | null>>(); const track = (effects: Map, sessionId: string, effect: E, done: Promise): Promise => { const list = effects.get(sessionId) ?? []; @@ -384,19 +384,43 @@ export function createSessionCatalog(db: () => Database, log?: Logger, scratchPi submit("effectiveThinkingLevel", sessionId, turnId, level).catch(err => reportWriteFailure(sessionId, err)); }, + // Settles once stored (#2850). Until then a pending exchange is still + // taken by the next prompt handed over, as when it committed inline. recordLocalExchange(sessionId, exchange, delivered) { - try { - saveLocalExchange(db(), sessionId, exchange, delivered); - return true; - } catch (err) { - reportWriteFailure(sessionId, err); - return false; - } + const done = submit("localExchange", sessionId, exchange, delivered, Date.now()); + if (!delivered) track(pendingExchanges, sessionId, exchange, done); + return done.then(() => true, (err: unknown) => { reportWriteFailure(sessionId, err); return false; }); }, + // Taken where the prompt is handed over, at once: the committed pending + // exchanges and those still committing, less any another prompt has + // already taken. Exactly these are then marked carried, after their + // records, on the write worker. takePendingLocalExchanges(sessionId) { try { - return takePendingLocalExchanges(db(), sessionId); + const claimed = takenExchanges.get(sessionId) ?? new Set(); + const seen = new Set(); + const taken: LocalExchange[] = []; + for (const record of loadLocalExchanges(db(), sessionId)) { + if (record.delivered || claimed.has(record.id) || seen.has(record.id)) continue; + seen.add(record.id); + const { seq: _seq, delivered: _delivered, ...exchange } = record; + taken.push(exchange); + } + for (const exchange of pendingExchanges.get(sessionId) ?? []) { + if (claimed.has(exchange.id) || seen.has(exchange.id)) continue; + seen.add(exchange.id); + taken.push(exchange); + } + if (!taken.length) return []; + const ids = taken.map(exchange => exchange.id); + for (const id of ids) claimed.add(id); + takenExchanges.set(sessionId, claimed); + void submit("exchangesDelivered", sessionId, ids, Date.now()).catch(err => reportWriteFailure(sessionId, err)).finally(() => { + for (const id of ids) claimed.delete(id); + if (!claimed.size && takenExchanges.get(sessionId) === claimed) takenExchanges.delete(sessionId); + }); + return taken; } catch (err) { // The prompt still goes out, without the figures: losing the turn // over them would be worse than the agent not seeing them. diff --git a/packages/ui-server/src/ws/session-profile-writes.ts b/packages/ui-server/src/ws/session-profile-writes.ts index dcf76089f6..2d2b32135c 100644 --- a/packages/ui-server/src/ws/session-profile-writes.ts +++ b/packages/ui-server/src/ws/session-profile-writes.ts @@ -3,11 +3,12 @@ * profile row references its catalog row, which commits on the `ui.db` write * worker once `session_info` names the session; binding a draft's preference * and recording the admitted profile commit there too, after it, at - * `synchronous=FULL`. A saved preference stays an app-connection write. + * `synchronous=FULL`, and so does a saved preference. */ import type { Database } from "bun:sqlite"; -import type { WriteProcedures } from "../db/write-worker.js"; +import type { WireError, WriteProcedures } from "../db/write-worker.js"; +import { SESSION_DRAFT_LIMITS } from "@schlessera/brain-ui-sdk/protocol"; import { isUsablePrincipal, resolvePrincipal } from "../db/principals.js"; export const url = import.meta.url; @@ -47,8 +48,46 @@ export function recordSessionProfile(db: Database, sessionId: string, origin: "i .run(sessionId, now, origin, actualProfile); } +/** + * A principal's saved preference for a session or a draft, checked and + * stored in one transaction. `policyProfiles` is the brain's policy profile + * for each origin, resolved where the request arrived. + */ +export function saveProfile(db: Database, principalId: string, kind: "session" | "draft", id: string, preference: Profile, + policyProfiles: Record<"interactive" | "voice", Profile>, now: number) { + db.transaction(() => { + authorizeProfile(db, principalId, now); + let origin: "interactive" | "voice" = "interactive"; + if (kind === "session") { + if (!db.query("SELECT 1 FROM sessions WHERE id = ?").get(id)) throw new SessionProfileError(404, "This chat is not known to the host."); + const row = db.query("SELECT origin FROM session_worker_profiles WHERE session_id = ?").get(id) as { origin: "interactive" | "voice" } | null; + origin = row?.origin ?? "interactive"; + } + if (origin === "voice") throw new SessionProfileError(409, "Voice uses this brain's policy."); + if (preference === "governed" && policyProfiles[origin] === "strict") throw new SessionProfileError(409, "Set by this brain's policy."); + if (kind === "session") { + db.query("INSERT INTO session_worker_profiles (session_id, preference, updated_at, updated_by) VALUES (?, ?, ?, ?) ON CONFLICT(session_id) DO UPDATE SET preference = excluded.preference, updated_at = excluded.updated_at, updated_by = excluded.updated_by") + .run(id, preference, now, principalId); + } else { + // Expired principals cannot use their old preferences; bounded cleanup does not evict a live confirmed choice. + db.query("DELETE FROM draft_worker_profiles WHERE principal_id NOT IN (SELECT id FROM principals WHERE revoked_at IS NULL AND expires_at > ?)").run(now); + if (!draftProfile(db, principalId, id) && (db.query("SELECT COUNT(*) AS n FROM draft_worker_profiles").get() as { n: number }).n >= SESSION_DRAFT_LIMITS.maxDrafts) + throw new SessionProfileError(507, "The host cannot keep another draft preference yet."); + db.query("INSERT INTO draft_worker_profiles (principal_id, draft_id, preference, updated_at) VALUES (?, ?, ?, ?) ON CONFLICT(principal_id, draft_id) DO UPDATE SET preference = excluded.preference, updated_at = excluded.updated_at") + .run(principalId, id, preference, now); + } + }).immediate(); +} + +/** A refusal keeps its class and status across the worker. */ +export function reviveProfileError(error: WireError): Error | undefined { + return error.name === "SessionProfileError" ? new SessionProfileError(error.fields.status as SessionProfileError["status"], error.message) : undefined; +} + export function setup(db: Database): WriteProcedures { return { + save: (principalId: string, kind: "session" | "draft", id: string, preference: Profile, policyProfiles: Record<"interactive" | "voice", Profile>, now: number) => + saveProfile(db, principalId, kind, id, preference, policyProfiles, now), bind: (principalId: string, draftId: string, sessionId: string, now: number) => bindDraftProfile(db, principalId, draftId, sessionId, now), record: (sessionId: string, origin: "interactive" | "voice", actualProfile: Profile, now: number) => recordSessionProfile(db, sessionId, origin, actualProfile, now), }; diff --git a/packages/ui-server/src/ws/session-profiles.ts b/packages/ui-server/src/ws/session-profiles.ts index 0d098af94d..d03f123f26 100644 --- a/packages/ui-server/src/ws/session-profiles.ts +++ b/packages/ui-server/src/ws/session-profiles.ts @@ -1,7 +1,7 @@ /** Authenticated preferences only; admission independently enforces its frozen policy. */ import { scopeLimits } from "../brain/turn-scope.js"; import type { Database } from "bun:sqlite"; -import { SESSION_DRAFT_LIMITS, type SessionWorkerProfile } from "@schlessera/brain-ui-sdk/protocol"; +import type { SessionWorkerProfile } from "@schlessera/brain-ui-sdk/protocol"; import type { GovernedTurns } from "../brain/governed-turn.js"; import { procedureCaller, type SqliteWriter } from "../db/write-worker.js"; import * as writes from "./session-profile-writes.js"; @@ -22,7 +22,7 @@ export function createSessionProfiles(db: Database, turns: GovernedTurns, budget */ options: { writer?: SqliteWriter; catalog?: { hasSession?(sessionId: string): boolean; settled?(): Promise } } = {}) { const authorize = (principalId: string) => authorizeProfile(db, principalId, Date.now()); - const submit = procedureCaller(db, writes, undefined, options.writer); + const submit = procedureCaller(db, writes, undefined, options.writer, writes.reviveProfileError); // Each session's submitted, unsettled bind and record, in order. Both are // idempotent over the row they produce, so a read applies them to the // committed row whether or not they have landed yet. @@ -85,26 +85,13 @@ export function createSessionProfiles(db: Database, turns: GovernedTurns, budget return { read, settled, - // A saved preference still commits here, after what it reads has settled. + // A saved preference commits on the write worker too (#2850), after + // what it reads has settled; the answer is read once it has. async save(principalId: string, kind: "session" | "draft", id: string, preference: Profile) { await settled(); - return db.transaction(() => { - const current = read(principalId, kind, id); - if (current.origin === "voice") throw new SessionProfileError(409, "Voice uses this brain's policy."); - if (preference === "governed" && current.policyProfile === "strict") throw new SessionProfileError(409, "Set by this brain's policy."); - if (kind === "session") { - db.query("INSERT INTO session_worker_profiles (session_id, preference, updated_at, updated_by) VALUES (?, ?, ?, ?) ON CONFLICT(session_id) DO UPDATE SET preference = excluded.preference, updated_at = excluded.updated_at, updated_by = excluded.updated_by") - .run(id, preference, Date.now(), principalId); - } else { - // Expired principals cannot use their old preferences; bounded cleanup does not evict a live confirmed choice. - db.query("DELETE FROM draft_worker_profiles WHERE principal_id NOT IN (SELECT id FROM principals WHERE revoked_at IS NULL AND expires_at > ?)").run(Date.now()); - if (!draft(principalId, id) && (db.query("SELECT COUNT(*) AS n FROM draft_worker_profiles").get() as { n: number }).n >= SESSION_DRAFT_LIMITS.maxDrafts) - throw new SessionProfileError(507, "The host cannot keep another draft preference yet."); - db.query("INSERT INTO draft_worker_profiles (principal_id, draft_id, preference, updated_at) VALUES (?, ?, ?, ?) ON CONFLICT(principal_id, draft_id) DO UPDATE SET preference = excluded.preference, updated_at = excluded.updated_at") - .run(principalId, id, preference, Date.now()); - } - return read(principalId, kind, id); - }).immediate(); + const policyProfiles = { interactive: turns.facts("interactive").policyProfile, voice: turns.facts("voice").policyProfile }; + await submit("save", principalId, kind, id, preference, policyProfiles, Date.now()); + return read(principalId, kind, id); }, preference(principalId: string, sessionId?: string, draftId?: string): Profile | undefined { authorize(principalId); diff --git a/packages/ui-server/tests/app-writes-offloop.test.ts b/packages/ui-server/tests/app-writes-offloop.test.ts index 27332d2ff5..882c92ce60 100644 --- a/packages/ui-server/tests/app-writes-offloop.test.ts +++ b/packages/ui-server/tests/app-writes-offloop.test.ts @@ -18,7 +18,8 @@ import { watchWriteSites } from "./helpers/write-sites.js"; import type { Logger } from "@opentelemetry/api-logs"; import { createPrincipal } from "../src/db/principals.js"; import { createSessionCatalog } from "../src/ws/session-catalog.js"; -import { createSessionProfiles } from "../src/ws/session-profiles.js"; +import { createSessionProfiles, SessionProfileError } from "../src/ws/session-profiles.js"; +import { expectRejection } from "./helpers/expect-rejection.js"; import { createPolicyProposals } from "../src/brain/policy-proposal.js"; import { createSessionRoutes } from "../src/routes/sessions.js"; import { createStaticBackendRegistry } from "../src/agent/backend.js"; @@ -297,3 +298,59 @@ test("only a session's newest acceptance orders or unorders it", () => { settle[0]!(true); expect(host.coordinator.unorderedSessions.has("troy"), "WORK_OLDER_ACCEPTANCE_DOES_NOT_REORDER").toBe(true); }); + +test("a local exchange commits on the write worker; still pending, the next prompt takes it once and marks it carried after its record (#2850)", async () => { + const f = fixture(); + try { + const catalog = createSessionCatalog(() => f.db, quietLog, undefined, { pricing: { resolve: () => null }, store: createActivityStore(f.db, { writer: "odysseus" }), writer: f.writer }); + const sites = await watchWriteSites(f.db, f.writer, ["local_exchanges"]); + const exchange = { id: "stats-ithaca", command: "stats", prompt: "Stats", context: "Documents: 3", answer: [] } as never; + const held = f.writer.channel(probe, { table: "odysseus_log" }); await held.ready; + const slow = held.call("slow", "Calypso", 300); + const recorded = catalog.recordLocalExchange!("ithaca", exchange, false); + expect(f.db.query("SELECT COUNT(*) AS n FROM local_exchanges").get(), "EXCHANGE_NOT_COMMITTED_ON_LOOP").toEqual({ n: 0 }); + expect(catalog.takePendingLocalExchanges!("ithaca").map(item => item.id), "EXCHANGE_PENDING_TAKEN_BY_NEXT_PROMPT").toEqual(["stats-ithaca"]); + expect(catalog.takePendingLocalExchanges!("ithaca"), "EXCHANGE_TAKEN_ONCE").toEqual([]); + await slow; + expect(await recorded, "EXCHANGE_RECORD_SETTLES_STORED").toBe(true); + await catalog.settled!(); + expect(f.db.query("SELECT delivered_at IS NOT NULL AS carried FROM local_exchanges WHERE exchange_id = 'stats-ithaca'").get(), "EXCHANGE_MARKED_CARRIED_AFTER_ITS_RECORD").toEqual({ carried: 1 }); + expect(catalog.takePendingLocalExchanges!("ithaca"), "EXCHANGE_CARRIED_NOT_TAKEN_AGAIN").toEqual([]); + expect(new Set(await sites.log()), "EXCHANGE_WRITTEN_ON_WORKER").toEqual(new Set(["local_exchanges"])); + } finally { await f.close(); } +}); + +test("a saved worker preference commits on the write worker, and its refusals keep their class and status (#2850)", async () => { + const f = fixture(); + try { + const athena = createPrincipal(f.db, { authMethod: "password", label: "Athena device", ttlSeconds: 3600 }); + f.db.query("INSERT INTO session_worker_profiles (session_id, preference, updated_at, updated_by, origin) VALUES ('ithaca', 'governed', 0, 'host', 'voice')").run(); + f.db.query("INSERT INTO sessions(id,title,created_at,last_active_at) VALUES ('troy','Troy',0,0)").run(); + const sites = await watchWriteSites(f.db, f.writer, ["session_worker_profiles", "draft_worker_profiles"]); + const catalog = createSessionCatalog(() => f.db, quietLog, undefined, { pricing: { resolve: () => null }, store: createActivityStore(f.db, { writer: "odysseus" }), writer: f.writer }); + const profiles = createSessionProfiles(f.db, profileFacts, undefined, undefined, { writer: f.writer, catalog }); + expect((await profiles.save(athena.id, "session", "troy", "strict")).preference, "PROFILE_SAVE_ANSWERS_COMMITTED_ROW").toBe("strict"); + expect((await profiles.save(athena.id, "draft", "troy-draft", "strict")).preference, "PROFILE_DRAFT_SAVE_COMMITS").toBe("strict"); + const refusal = await expectRejection(profiles.save(athena.id, "session", "ithaca", "strict"), SessionProfileError, "PROFILE_REFUSAL_CLASS_KEPT") as SessionProfileError; + expect(refusal.status, "PROFILE_REFUSAL_STATUS_KEPT").toBe(409); + expect(new Set(await sites.log()), "PROFILE_SAVES_WRITTEN_ON_WORKER").toEqual(new Set(["session_worker_profiles", "draft_worker_profiles"])); + } finally { await f.close(); } +}); + +test("a replayed history waits for the classified blocks submitted before it (#2850)", async () => { + const f = fixture(); + try { + const catalog = createSessionCatalog(() => f.db, quietLog, undefined, { pricing: { resolve: () => null }, store: createActivityStore(f.db, { writer: "odysseus" }), writer: f.writer }); + let release!: () => void; + const pending = new Promise(done => { release = done; }); + const classifier = { enabled: true, settled: () => pending, attach: (_sessionId: string, messages: never[]) => messages } as never; + const host = new WsHost({ registry: createStaticBackendRegistry([makeFakeBackend({ id: "claude" })], "claude"), catalog, classifier, maxConcurrentSessions: () => 3 }); + let settled = false; + const waiting = host.historySettled().then(() => { settled = true; }); + await Bun.sleep(50); + expect(settled, "HISTORY_WAITS_FOR_PENDING_CLASSIFIED_BLOCKS").toBe(false); + release(); + await waiting; + expect(settled).toBe(true); + } finally { await f.close(); } +}); diff --git a/packages/ui-server/tests/classification.test.ts b/packages/ui-server/tests/classification.test.ts index dbfa857d77..98011554d5 100644 --- a/packages/ui-server/tests/classification.test.ts +++ b/packages/ui-server/tests/classification.test.ts @@ -15,6 +15,9 @@ import type { ServerMessage, SessionHistoryMessage } from "@schlessera/brain-ui- import { CONFIDENCE, questionsFor } from "../../ui-sdk/src/classification/catalogue.js"; import { planClassification } from "@schlessera/brain-ui-sdk/internal"; import { createUiDb } from "../src/db/client"; +import { openSqliteWriter } from "../src/db/write-worker"; +import * as probe from "./fixtures/write-worker-probe"; +import { watchWriteSites } from "./helpers/write-sites"; import { JEV_ENDPOINT, JEV_TIMEOUT_MS, @@ -472,6 +475,28 @@ describe("the confidence record", () => { ]); }); + test("with a write worker, a pass commits its blocks and confidence there, and settles once its blocks have (#2850)", async () => { + const writer = openSqliteWriter(db); + try { + const sites = await watchWriteSites(db, writer, ["message_blocks", "classification_confidence"]); + // A failed write is only a log line: record them so it surfaces with its reason. + const warnings: unknown[] = []; + const log = { emit: (record: { attributes?: Record }) => { warnings.push(record.attributes?.["error.message"]); } } as unknown as typeof silentLog; + const c = createTurnClassifier({ jev: { breaker: () => ({ open: false, consecutiveFailures: 0, retryAt: null }), ...answering(GOOD_ANSWERS) }, + db: () => db, log, writer }); + const held = writer.channel(probe, { table: "odysseus_log" }); await held.ready; + const slow = held.call("slow", "Calypso", 300); + const blocks = await c.run("s-worker", [TEXT]); + expect(warnings, "CLASSIFY_WORKER_WRITES_DID_NOT_FAIL").toEqual([]); + expect(blocks, "CLASSIFY_WORKER_PASS_SWAPS").toHaveLength(1); + expect(db.query("SELECT COUNT(*) AS n FROM message_blocks WHERE session_id = 's-worker'").get(), "CLASSIFY_BLOCKS_COMMITTED_BEFORE_RUN_SETTLES").toEqual({ n: 1 }); + await slow; await c.settled!(); + expect(warnings, "CLASSIFY_WORKER_WRITES_DID_NOT_FAIL").toEqual([]); + expect(rows("s-worker"), "CLASSIFY_CONFIDENCE_COMMITTED_ON_WORKER").toHaveLength(3); + expect(new Set(await sites.log()), "CLASSIFY_TABLES_WRITTEN_ON_WORKER").toEqual(new Set(["message_blocks", "classification_confidence"])); + } finally { await writer.close(); } + }); + test("a candidate kept because its answer was under the line records the number", async () => { // This is what `kept` used to be: a count with nothing behind it. const low = { "p0c0.shape": { type: "choice", choice: "comparison", probabilities: {}, confidence: 0.58 } }; diff --git a/packages/ui-server/tests/ws-legacy-budget-replay.test.ts b/packages/ui-server/tests/ws-legacy-budget-replay.test.ts index 7cb97c4ebc..13bec7f6b7 100644 --- a/packages/ui-server/tests/ws-legacy-budget-replay.test.ts +++ b/packages/ui-server/tests/ws-legacy-budget-replay.test.ts @@ -54,7 +54,7 @@ for (const adapter of ["claude", "pi"] as const) { snapshot = await readFile(nativePath, "utf8"); load = () => getPiHistory(brain, sessionId, sessionDir); } - expect(catalog.recordLocalExchange!(sessionId, exchange, true)).toBe(true); + expect(await catalog.recordLocalExchange!(sessionId, exchange, true)).toBe(true); catalog.recordMessageSource!(sessionId, text, "voice-dictate", { thinkingLevel: "high", turnId: "legacy-turn", files }); const backendHistory = await load(); expect(backendHistory[0]!.content).toBe(prompt); diff --git a/packages/ui-server/tests/ws-turn-failure-replay.test.ts b/packages/ui-server/tests/ws-turn-failure-replay.test.ts index 30c78440a1..5b4e428921 100644 --- a/packages/ui-server/tests/ws-turn-failure-replay.test.ts +++ b/packages/ui-server/tests/ws-turn-failure-replay.test.ts @@ -244,7 +244,7 @@ for (const id of ["claude", "pi"] as const) describe(`${id}: live failure surviv const first: TurnFailure = { errorClass: "invalid_request", status: 400, message: "API Error: 400 Request refused." }; const second: TurnFailure = { ...first, errorClass: "model_not_found" }; await h.send(first); - expect(testHost().catalog.recordLocalExchange?.("s1", { + expect(await testHost().catalog.recordLocalExchange?.("s1", { id: "local-stats", command: "stats", prompt: "Stats", context: "Documents: 3", answer: [{ kind: "tiles", source: "corpus", tiles: [{ label: "Documents", value: "3" }] }], }, false)).toBe(true);