From 1a9d1e2109ebd6ab46dabf26f5293f6d106fdb87 Mon Sep 17 00:00:00 2001 From: Alain Schlesser Date: Sun, 25 Oct 2026 16:31:06 +0100 Subject: [PATCH 1/4] fix(ui-server): commit the session catalog on the write worker The hosted server no longer commits the session catalog on its app connection (#2850). The session row, accepted work, turn boundaries, message sources, turn failures, the state room, Claude usage accounting, handoff links, labels and session costs commit on the ui.db write worker at synchronous=FULL, in submission order with the journal's, Activity's, governance, scope and budget writes. So do the two session_info writes that reference the new row: binding a draft's worker profile and recording the admitted one, and binding the turn's policy proposals. Synchronous reads on the turn path apply every pending write's own effect to the committed row, so a write counts from its submission as it did when it committed inline; each effect stops counting when its write settles. History joins, the session list and history routes, handoff sources, the policy proposal route and retry reads await settled() first. An acceptance's revision is assigned where it is accepted, never reissued after a failure, and only the newest acceptance orders the session. Retained retries and local exchanges stay on the app connection. Refs #2850 Co-Authored-By: Claude Opus 5.5 --- .changeset/session-catalog-off-event-loop.md | 5 + docs/decisions/answer-delivery.md | 2 +- docs/decisions/async-collaboration.md | 2 +- docs/decisions/governed-writable-workers.md | 10 +- docs/decisions/http-api-boundary.md | 8 +- docs/decisions/session-handoff.md | 2 +- docs/decisions/session-principals.md | 6 +- docs/decisions/sqlite-write-worker.md | 44 ++- docs/decisions/voice-permission.md | 14 +- .../live-conversation-investigation.md | 4 +- .../plans/async-collaboration-requirements.md | 6 +- docs/plans/async-collaboration.md | 6 +- docs/plans/navigation-recovery.md | 6 +- .../ui-server/src/activity/store-writes.ts | 21 +- packages/ui-server/src/app.ts | 13 +- .../src/brain/policy-proposal-writes.ts | 20 ++ .../ui-server/src/brain/policy-proposal.ts | 14 +- .../ui-server/src/routes/policy-proposals.ts | 5 +- .../ui-server/src/routes/session-profiles.ts | 2 +- packages/ui-server/src/routes/sessions.ts | 8 + packages/ui-server/src/ws/bridge.ts | 10 +- packages/ui-server/src/ws/dispatch.ts | 7 + packages/ui-server/src/ws/handoff.ts | 20 +- packages/ui-server/src/ws/host.ts | 12 + packages/ui-server/src/ws/message-sources.ts | 5 +- packages/ui-server/src/ws/recovery.ts | 19 +- .../src/ws/session-catalog-writes.ts | 165 +++++++++ packages/ui-server/src/ws/session-catalog.ts | 323 ++++++++---------- .../src/ws/session-profile-writes.ts | 55 +++ packages/ui-server/src/ws/session-profiles.ts | 92 +++-- packages/ui-server/src/ws/session-work.ts | 15 +- .../tests/app-writes-offloop.test.ts | 161 +++++++++ .../tests/session-routes-handoff.test.ts | 6 +- packages/ui-server/tests/ws-handoff.test.ts | 2 +- .../ui-server/tests/ws-scope-caps.test.ts | 6 +- 35 files changed, 821 insertions(+), 275 deletions(-) create mode 100644 .changeset/session-catalog-off-event-loop.md create mode 100644 packages/ui-server/src/brain/policy-proposal-writes.ts create mode 100644 packages/ui-server/src/ws/session-catalog-writes.ts create mode 100644 packages/ui-server/src/ws/session-profile-writes.ts diff --git a/.changeset/session-catalog-off-event-loop.md b/.changeset/session-catalog-off-event-loop.md new file mode 100644 index 0000000000..586a677dfd --- /dev/null +++ b/.changeset/session-catalog-off-event-loop.md @@ -0,0 +1,5 @@ +--- +"@schlessera/brain-ui-server": patch +--- + +Changed: the hosted server commits the session catalog on its SQLite write worker at `synchronous=FULL`, in submission order, instead of on the event loop (#2850): the session row, accepted work and turn boundaries, message sources, turn failures, the state room, Claude usage accounting, handoff links, labels and session costs, and the worker-profile and policy-proposal binds a new session makes. Reads on the turn path see a write from its submission; the session list, history replays and retry reads wait for the writes submitted before them. diff --git a/docs/decisions/answer-delivery.md b/docs/decisions/answer-delivery.md index 33518b7313..6b8abd14d7 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:641`). Before this, a resume replaced +`packages/ui-server/src/ws/dispatch.ts:648`). Before this, a resume replaced the transcript and left only a history card behind, which was the second symptom. diff --git a/docs/decisions/async-collaboration.md b/docs/decisions/async-collaboration.md index ea6afd8ed6..ed1c8eaf82 100644 --- a/docs/decisions/async-collaboration.md +++ b/docs/decisions/async-collaboration.md @@ -204,7 +204,7 @@ permission and unwinds the turn. Already completed side effects are not undone by an abort. Attempt staging and cleanup therefore need idempotent recovery, with filesystem compensation outside SQLite transactions. Ordinary permission parking still exists (`requestPermission: (req) => {`, -`packages/ui-server/src/ws/bridge.ts:354-447`); it is not a durable Action store. +`packages/ui-server/src/ws/bridge.ts:358-451`); it is not a durable Action store. Resolution validates the stored effect again, checks current authority, records one resolution and applies the guarded state transition in one transaction. diff --git a/docs/decisions/governed-writable-workers.md b/docs/decisions/governed-writable-workers.md index 112e845840..e71105735a 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:12-89`; +(`createSessionProfiles`, `packages/ui-server/src/ws/session-profiles.ts:16-132`; `begin`, `packages/ui-server/src/brain/governed-turn.ts:142-306`). **Where a turn falls back to `strict`.** The launcher refuses to start @@ -163,7 +163,7 @@ still holds. **Voice.** Ruling 1 makes voice `governed` too. Voice keeps its narrower membership (`VOICE_ALLOWED_TOOLS`, `packages/ui-backend-claude/src/tool-policy.ts:120`): no Bash, no raw `Write`/`Edit`, no archive, and approval returns false -(`const apply = createBrainApplication`, `packages/ui-server/src/ws/bridge.ts:658-663`). +(`const apply = createBrainApplication`, `packages/ui-server/src/ws/bridge.ts:662-667`). A voice turn therefore writes only through curated capture and update, now with a journal behind it. The writable mount gives voice no tool it lacked. Pi still refuses voice turns. @@ -1072,7 +1072,7 @@ principal authority remain the caller's responsibility. The policy publisher creates, writes, fsyncs and removes its operation-owned staging name entirely inside that synchronous callback, between the two serialized observations (`await options.observeUserPolicy`, -`packages/ui-server/src/brain/policy-proposal.ts:198-225`). First publication +`packages/ui-server/src/brain/policy-proposal.ts:210-237`). First publication also removes the temporary hardlink before the final observation. No filename prefix is ignored; a similarly named real agent change is still retained. Failed staging preserves the original policy, while failed recording after @@ -1531,7 +1531,7 @@ The brain's git history stays the durable record of its content. - **The Activity recorder and its rollups** price those events under the [cost-tracking rules](cost-tracking.md). An unknown price is never counted as zero. The session total that counts unknown as zero - (`msg.costUsd ?? 0`, `packages/ui-server/src/ws/session-catalog.ts:368`) + (`msg.costUsd ?? 0`, `packages/ui-server/src/ws/session-catalog.ts:375`) must not feed any budget. **Enforcement:** @@ -2429,7 +2429,7 @@ every turn without a proven recipient: - **The handoff summary.** Drafting a handoff summary refuses when the policy has provider rules and the source chat's lineage does not cover every provider folder, before its history is read (`unprovenHistory`, - `packages/ui-server/src/ws/handoff.ts:303-306`). The run has no proven + `packages/ui-server/src/ws/handoff.ts:309-312`). The run has no proven recipient, so every provider folder is denied to it. A policy with a type selector always refuses: the type's folders follow the current taxonomy, which this check does not translate, and a remembered translation may be diff --git a/docs/decisions/http-api-boundary.md b/docs/decisions/http-api-boundary.md index dd8ae50547..50a22b568c 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:857`), the API guard follows -public ceremonies (`authGuard`, `packages/ui-server/src/app.ts:878`), and the -WebSocket admission is separate (`"/ws"`, `packages/ui-server/src/app.ts:1067`). +(`createHealthRoutes`, `packages/ui-server/src/app.ts:861`), the API guard follows +public ceremonies (`authGuard`, `packages/ui-server/src/app.ts:882`), and the +WebSocket admission is separate (`"/ws"`, `packages/ui-server/src/app.ts:1072`). Static serving is conditional (`if (options.staticRoot)`, -`packages/ui-server/src/app.ts:1087-1108`). That order establishes access; +`packages/ui-server/src/app.ts:1092-1113`). 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-handoff.md b/docs/decisions/session-handoff.md index 2028cdf9f2..8d860445d9 100644 --- a/docs/decisions/session-handoff.md +++ b/docs/decisions/session-handoff.md @@ -160,7 +160,7 @@ its distinct retry treatment. Nothing archives, purges or automatically sends. The notice's presentation and actions are (`StateRoomNotice`, `packages/ui-react/src/components/chat/state-room-notice.tsx:22-100`); the host keeps the measurement and first quiet event (`if (msg.type === "result") {`, -`packages/ui-server/src/ws/bridge.ts:275-285`). +`packages/ui-server/src/ws/bridge.ts:279-289`). ### Historical state directories remain untouched diff --git a/docs/decisions/session-principals.md b/docs/decisions/session-principals.md index acec1943d1..b5639d1e08 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:863` | mounted **before** the guard at `authGuard(authMode`, `app.ts:878`; logout guards itself (`resolveCookiePrincipal(c, auth, deps.db)`, `auth.ts:808`) | -| Passkey management | `passkeyManagementRoutes`, `app.ts:888` | mounted **after** the guard ("Mount AFTER the auth guard") | -| WS guard | `"/ws"`, `app.ts:1067-1084` | origin → `isWsAuthorized` (returns a boolean, `auth.ts:221-237`) → capacity → upgrade | +| Auth routes | `authRoutes`, `app.ts:867` | mounted **before** the guard at `authGuard(authMode`, `app.ts:882`; logout guards itself (`resolveCookiePrincipal(c, auth, deps.db)`, `auth.ts:808`) | +| Passkey management | `passkeyManagementRoutes`, `app.ts:892` | mounted **after** the guard ("Mount AFTER the auth guard") | +| WS guard | `"/ws"`, `app.ts:1072-1089` | 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 9cc40d36ec..04c0d6d844 100644 --- a/docs/decisions/sqlite-write-worker.md +++ b/docs/decisions/sqlite-write-worker.md @@ -87,11 +87,13 @@ thread waits for it. ## The exceptions -- **A write inside the caller's transaction runs there.** The session catalog - writes title and unstreamed-usage spans inside its own transaction on the app - connection. The off-loop Activity store sees `db.inTransaction` and runs that - write synchronously on the caller's connection, so it stays atomic with the - caller's rows. That commit is the caller's, on the event loop, as before. +- **A write inside the caller's transaction runs there.** The off-loop + Activity store sees `db.inTransaction` and runs such a write synchronously on + the caller's connection, so it stays atomic with the caller's rows. Claude + usage accounting, which writes title and unstreamed-usage spans inside its + own transaction, now runs that whole transaction on the worker, with a store + on the worker's connection (`workerActivity`, + `packages/ui-server/src/activity/store-writes.ts`). - **The in-memory database has no file** for a second connection. Its writer runs the same procedures inline, in the same order, behind the same promises. - **Rollup pricing in the worker** reads the same on-disk table and bundled @@ -121,9 +123,37 @@ thread waits for it. at once; its warnings and any block follow the commit. A relayed provider request is admitted only after every budget write submitted before it has committed, and a turn's outcome waits for its budget writes. +- **The session catalog followed** (#2850). The session row, accepted work, + turn boundaries, message sources, turn failures, the state room, Claude + usage accounting, handoff links, labels and session costs commit on the + worker (`packages/ui-server/src/ws/session-catalog-writes.ts`), and so do + the two writes `session_info` makes that reference the new row: binding a + draft's worker profile and recording the admitted one + (`packages/ui-server/src/ws/session-profile-writes.ts`), and binding the + turn's policy proposals (`packages/ui-server/src/brain/policy-proposal-writes.ts`). + The worker's order puts each after the row it references. Readers see the + catalog's writes in two ways. A synchronous read on the turn path (whether + a session exists, its backend, provider and label, its latest accepted + work, a session's worker profile and whether a draft is still unconsumed) + answers from the committed row with every pending write's own effect + applied, so a write counts from its submission, as it did when it + committed inline. Each effect stops counting when its write settles, so a + failed one stops at once. A reader that joins committed rows onto history + (`publishHistory`, the session list and history routes, handoff sources, + the policy proposal route) or reads retained retries first awaits + `settled()` (`historySettled`, `packages/ui-server/src/ws/host.ts`). An + 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 + 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, - the session catalog and message sources, the Activity - notifier and push ticks, auth sessions, inbox and settings. Each commits at + 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 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 d70b6f98ec..99bedf6705 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:439-445`) +(the block computing `remembers`, `packages/ui-server/src/ws/dispatch.ts:446-452`) 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,8 +405,8 @@ 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:439-445`, and the lookup computing -`remembered`, `ws/bridge.ts:364-384`), and 192 of 192 measured +(the block computing `remembers`, `dispatch.ts:446-452`, and the lookup computing +`remembered`, `ws/bridge.ts:368-388`), and 192 of 192 measured approvals were kind `command`. ### When the announcement actually fires @@ -463,7 +463,7 @@ the request kind, and it patches the span (`onApprovalDecision`, `packages/ui-server/src/activity/recorder.ts:674-689`), fed from the bridge's `recorded()` wrapper -(`const recorded`, `packages/ui-server/src/ws/bridge.ts:406-421`). +(`const recorded`, `packages/ui-server/src/ws/bridge.ts:410-425`). One thing is missing and is a follow-up: **the event does not record the modality.** A denial decided by a phrase a microphone heard and one decided by a @@ -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:486-509`) with a message naming the +`packages/ui-server/src/ws/dispatch.ts:493-516`) with a message naming the phrase that produced it. ## Containment: shared with #51, deliberately not identical @@ -608,9 +608,9 @@ matters because a reader cannot otherwise tell a live hazard from a fixed one. purpose, so a grant that exists and is deliberately not applied can be logged — but refuses to answer from it or add to it for a tool the enforced allowlist left out (the lookup computing `remembered`, - `packages/ui-server/src/ws/bridge.ts:364-384`, + `packages/ui-server/src/ws/bridge.ts:368-388`, and the block computing `remembers`, - `packages/ui-server/src/ws/dispatch.ts:439-445`). + `packages/ui-server/src/ws/dispatch.ts:446-452`). 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 7123a7b143..6fc51740f2 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:399-422`). +(`if (msg.channel === "voice")`, `packages/ui-server/src/ws/dispatch.ts:406-429`). 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:486-509`). The +(`case "tool_denial"`, `packages/ui-server/src/ws/dispatch.ts:493-516`). 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 4c9d09c85b..8a949d3015 100644 --- a/docs/plans/async-collaboration-requirements.md +++ b/docs/plans/async-collaboration-requirements.md @@ -43,7 +43,7 @@ session. Three consequences: 1. **Background work cannot ask.** Cron runs (`sync`, `validate`, `maintain`, module jobs) have no human attached. `bridge.requestPermission` parks a promise that nobody will - resolve (`requestPermission: (req) => {`, `packages/ui-server/src/ws/bridge.ts:354-448`), so autonomous work is confined to + resolve (`requestPermission: (req) => {`, `packages/ui-server/src/ws/bridge.ts:358-452`), so autonomous work is confined to whatever is pre-approved, and anything requiring judgment is not attempted. 2. **Inbound material has no path.** A forwarded email, a shared link, a captured note has nowhere to land that the agent will act on later. The PWA share target stages into @@ -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:878`). + (`app.use("/api/*", authGuard(`, `packages/ui-server/src/app.ts:882`). 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 @@ -263,7 +263,7 @@ Stated before the requirements because five of them derive from it. threshold (~20s of the 30s lock budget). Below it nothing yields; above it the holder checkpoints/unwinds/releases. Test both edges and independent paths. - R26. **Abort-and-redo needs a checkpoint primitive.** `requestPermission` parks a bare - promise (`return new Promise((resolve) => {`, `packages/ui-server/src/ws/bridge.ts:398-456`) — while blocked on it the model cannot write anything, so + promise (`return new Promise((resolve) => {`, `packages/ui-server/src/ws/bridge.ts:402-460`) — while blocked on it the model cannot write anything, so "writes its findings, then aborts" has nowhere to run. The autonomous bridge must, in one server-side step: capture the checkpoint, create the Action and block the item, unwind without a live approval promise, preserving the tested timeout path's abort-then-drain order (`abortController.abort();`, `packages/ui-server/src/ws/run-session.ts:393-399`). Aborting does **not** undo completed tool side diff --git a/docs/plans/async-collaboration.md b/docs/plans/async-collaboration.md index 5e287e2108..4b4907a7dd 100644 --- a/docs/plans/async-collaboration.md +++ b/docs/plans/async-collaboration.md @@ -28,7 +28,7 @@ than last. ## Problem Frame Collaboration is synchronous-only: background work cannot ask (`requestPermission` parks a -promise nobody resolves — (`requestPermission: (req) => {`, `packages/ui-server/src/ws/bridge.ts:354-448`)), inbound material +promise nobody resolves — (`requestPermission: (req) => {`, `packages/ui-server/src/ws/bridge.ts:358-452`)), inbound material has no path that survives until the user is present, and no decision accumulates into standing authority. See origin for the full frame. @@ -131,10 +131,10 @@ 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:878`) follows public routes; +- **Auth mounting:** (`app.use("/api/*", authGuard(`, `packages/ui-server/src/app.ts:882`) follows public routes; (`export function authGuard(`, `packages/ui-server/src/middleware/auth.ts:189-249`) 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:354-448`). Timeout unwind is +- **Permission parking:** (`requestPermission: (req) => {`, `packages/ui-server/src/ws/bridge.ts:358-452`). Timeout unwind is (`abortController.abort();`, `packages/ui-server/src/ws/run-session.ts:393-399`). Durable escalation must checkpoint before unwinding; the existing ordinary bridge does not do that. - **Tool enforcement:** (`const enforcementHook: HookCallback`, `packages/ui-backend-claude/src/permission-hooks.ts:200-229`) closes measured bypasses. diff --git a/docs/plans/navigation-recovery.md b/docs/plans/navigation-recovery.md index b3123ef825..43b52de0a2 100644 --- a/docs/plans/navigation-recovery.md +++ b/docs/plans/navigation-recovery.md @@ -385,9 +385,9 @@ 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:621-675`). +- Host resume: (`case "session_resume": {`, `packages/ui-server/src/ws/dispatch.ts:628-682`). -- Catalog activity: (`persistSessionStub(sessionId, promptText, providerId, backendId) {`, `packages/ui-server/src/ws/session-catalog.ts:339-378`). +- Catalog activity: (`persistSessionStub(sessionId, promptText, providerId, backendId) {`, `packages/ui-server/src/ws/session-catalog.ts:369-376`). - Session summary shape: (`export interface ChatSession {`, `packages/ui-sdk/src/protocol.ts:1568-1584`). @@ -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:878`). +- API authorization: (`app.use("/api/*", authGuard(authMode, auth, db));`, `packages/ui-server/src/app.ts:882`). - 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/activity/store-writes.ts b/packages/ui-server/src/activity/store-writes.ts index 4c81924994..b74f1b885d 100644 --- a/packages/ui-server/src/activity/store-writes.ts +++ b/packages/ui-server/src/activity/store-writes.ts @@ -28,6 +28,21 @@ export interface ActivityWriteOptions { writer: string; pricing?: ActivityPricin * replaces it, so a rollup prices from the table the app last published. */ export function setup(db: Database, options: ActivityWriteOptions): WriteProcedures { + const { store, pricing } = workerActivity(db, options); + const procedures: WriteProcedures = {}; + for (const name of ACTIVITY_WRITE_METHODS) { + procedures[name] = ((...args: unknown[]) => { pricing?.(); return Reflect.apply(store[name] as (...values: unknown[]) => unknown, store, args); }) as WriteProcedures[string]; + } + return procedures; +} + +/** + * A store on the write connection, as `setup` builds it, for another + * procedure module whose transaction writes Activity rows too (#2850). + * `pricing`, present with a pricing configuration, reloads the shared table + * if the app has replaced it; call it before a write that prices. + */ +export function workerActivity(db: Database, options: ActivityWriteOptions): { store: ActivityStore; pricing?: () => RollupPricing } { const config = options.pricing; let pricing: RollupPricing | undefined, identity = ""; const current = (): RollupPricing => { @@ -38,11 +53,7 @@ export function setup(db: Database, options: ActivityWriteOptions): WriteProcedu return pricing; }; const store = createActivityStore(db, { writer: options.writer, ...(config ? { pricing: { resolve: (model, route) => current().resolve(model, route) } } : {}) }); - const procedures: WriteProcedures = {}; - for (const name of ACTIVITY_WRITE_METHODS) { - procedures[name] = ((...args: unknown[]) => { if (config) current(); return Reflect.apply(store[name] as (...values: unknown[]) => unknown, store, args); }) as WriteProcedures[string]; - } - return procedures; + return { store, ...(config ? { pricing: current } : {}) }; } /** diff --git a/packages/ui-server/src/app.ts b/packages/ui-server/src/app.ts index 265ed1bb58..6b62935556 100644 --- a/packages/ui-server/src/app.ts +++ b/packages/ui-server/src/app.ts @@ -611,7 +611,11 @@ export async function createApp(options: CreateAppOptions = {}): Promise host.turnTimeoutMs }); + // Catalog rows, and the profile rows and proposal binds that reference + // them, commit on the write worker (#2850). + const catalog = createSessionCatalog(() => db, dbLog, scratchPins, { pricing, store: activity.store, governance: activity.governanceWrites, writer, + pricingConfig: { brainPath: config.brainPath, enabled: config.pricing.enabled, ttlMs: config.pricing.ttlMs } }); + const sessionProfiles = createSessionProfiles(db, governedTurns, budgets, { hardTimeoutMs: () => host.turnTimeoutMs }, { writer, catalog }); const journal = createTurnJournal(db, { nativeManager, brainPath: config.brainPath, dbPath: config.dbPath, objectsRoot: journalRoot, writer, ...(referencePoolConfigured ? { referencePool: nativeManager, referenceCheckpoints: (() => { const owner = referenceWorktreeOwner(); @@ -621,7 +625,7 @@ export async function createApp(options: CreateAppOptions = {}): Promise journal.close()); // Startup recovery commits, or refuses boot, before any turn can run. await journal.ready; - const policyProposals = createPolicyProposals({ root: config.brainPath, db, + const policyProposals = createPolicyProposals({ root: config.brainPath, db, writer, lock: { withLock: (key, fn, admission) => host.workerLock.withLock(key, fn, admission) }, observeUserPolicy: (turnId, write) => journal.observeUserPolicy(turnId, write), recordEvent: (review, turn) => activity.governanceWrites.record({ kind: "policy_change", actor: "user", turnId: turn.turnId, sessionId: turn.sessionId(), model: null, profile: null, @@ -648,7 +652,7 @@ export async function createApp(options: CreateAppOptions = {}): Promise db, dbLog, scratchPins, { pricing, store: activity.store, governance: activity.governanceWrites }), + catalog, // Exact-operation approvals remain fail-closed until the full-v1 engine // wires current server authority. Dismiss/cancel/snooze need no grant. inboxTools: createInboxBridgeTools(db, intake, hygieneReview.resolver, config.brainPath), @@ -935,7 +939,8 @@ export async function createApp(options: CreateAppOptions = {}): Promise host.prepareHistory(id, messages) })); + app.route("/api", createSessionRoutes({ registry, db, labels: host.labels !== null, prepareHistory: (id, messages) => host.prepareHistory(id, messages), + settled: () => host.historySettled() })); app.route("/api", createJournalRoutes({ journal })); const storageHistory = createStorageHistory(db, config.brainPath); bootCleanup.push(() => storageHistory.close()); diff --git a/packages/ui-server/src/brain/policy-proposal-writes.ts b/packages/ui-server/src/brain/policy-proposal-writes.ts new file mode 100644 index 0000000000..52ea30ad39 --- /dev/null +++ b/packages/ui-server/src/brain/policy-proposal-writes.ts @@ -0,0 +1,20 @@ +/** + * A policy proposal joins its session off the event loop (#2850): the bind + * references the session's catalog row, which commits on the `ui.db` write + * worker once `session_info` names the session, so the bind commits there + * too, after it, at `synchronous=FULL`. A proposal recorded before the row + * exists keeps a null session until the bind lands. + */ +import type { Database } from "bun:sqlite"; + +import type { WriteProcedures } from "../db/write-worker.js"; + +export const url = import.meta.url; + +export function bindPolicyProposals(db: Database, turnId: string, sessionId: string) { + db.query("UPDATE policy_proposals SET session_id = ? WHERE turn_id = ? AND session_id IS NULL").run(sessionId, turnId); +} + +export function setup(db: Database): WriteProcedures { + return { bind: (turnId: string, sessionId: string) => bindPolicyProposals(db, turnId, sessionId) }; +} diff --git a/packages/ui-server/src/brain/policy-proposal.ts b/packages/ui-server/src/brain/policy-proposal.ts index cfa94586b0..660a20ee13 100644 --- a/packages/ui-server/src/brain/policy-proposal.ts +++ b/packages/ui-server/src/brain/policy-proposal.ts @@ -10,6 +10,8 @@ import { parseGovernancePolicy, GOVERNANCE_POLICY_PATH, GOVERNANCE_POLICY_MAX_BY import { BRAIN_LOCK_KEY, type KeyedLock } from "@schlessera/brain-ui-sdk/server"; import { PROPOSE_POLICY_CHANGE_INPUT_SCHEMA, type PolicyChangeInput, type PolicyChangeReview } from "@schlessera/brain-ui-sdk/tool-contracts"; import { admitHeld, AuthorityHeld, retryHeld, type AuthorityHold } from "./authority-hold.js"; +import { procedureCaller, type SqliteWriter } from "../db/write-worker.js"; +import * as proposalWrites from "./policy-proposal-writes.js"; const hash = (bytes: string | Uint8Array) => createHash("sha256").update(bytes).digest("hex"); class Refusal extends Error { constructor(readonly state: "stale" | "invalid" | "refused", message: string) { super(message); } } @@ -108,11 +110,15 @@ export function createPolicyProposals(options: { observeUserPolicy(turnId: string, approvedWrite: () => void): void | Promise; /** A returned promise is the event's commit (#2850); its failure is the receipt's. */ recordEvent(review: PolicyChangeReview, turn: PolicyProposalTurn): unknown; + /** Where a proposal's session bind commits: the app's write worker (#2850). Absent: on `db`, at once. */ + writer?: SqliteWriter; /** Deterministic race/revalidation fixtures, never wired to a worker. */ beforeCommit?(): void; validate?: typeof parseGovernancePolicy; }) { const db = options.db; + const bind = procedureCaller(db, proposalWrites, undefined, options.writer); + let bound: Promise = Promise.resolve(); // App startup also serves hosts whose brain has not been provisioned yet. // Keep existing roots canonical; a missing root is resolved at the operation, // never created by policy publication. @@ -146,7 +152,13 @@ export function createPolicyProposals(options: { return review ? { ...tool, policyChange: review } : tool; }) })); }, - bind(turnId: string, sessionId: string) { db.query("UPDATE policy_proposals SET session_id = ? WHERE turn_id = ? AND session_id IS NULL").run(sessionId, turnId); }, + bind(turnId: string, sessionId: string): Promise { + const done = bind("bind", turnId, sessionId); + bound = done.then(() => {}, () => {}); + return done; + }, + /** Settles once every bind submitted before it has settled: history joins proposals by session. */ + async settled() { let waiting: Promise; do { waiting = bound; await waiting; } while (waiting !== bound); }, async propose(turn: PolicyProposalTurn, requestId: string, raw: PolicyChangeInput): Promise { const authority = (waits = true) => { if (!turn.signal.aborted && !turn.isAuthorized() && waits && turn.hold?.active()) throw new AuthorityHeld(); diff --git a/packages/ui-server/src/routes/policy-proposals.ts b/packages/ui-server/src/routes/policy-proposals.ts index 46de8f78a6..fa5ede6154 100644 --- a/packages/ui-server/src/routes/policy-proposals.ts +++ b/packages/ui-server/src/routes/policy-proposals.ts @@ -3,12 +3,15 @@ import type { AppEnv } from "../app-env.js"; import type { WsHost } from "../ws/host.js"; /** Read-only same-request reconciliation. Approval is exclusively the existing WS authority route. */ export function createPolicyProposalRoutes(host: WsHost): Hono { - return new Hono().get("/policy-proposals/:turnId/:requestId", c => { + return new Hono().get("/policy-proposals/:turnId/:requestId", async c => { c.header("Cache-Control", "no-store"); const principal = c.get("principal"); if (!principal || !host.isPrincipalAuthorized(principal.id)) return c.json({ error: "Authentication required", authRequired: true }, 401); const { turnId, requestId } = c.req.param(); if (turnId.length > 200 || requestId.length > 200) return c.json({ error: "Invalid request identity" }, 400); + // A proposal joins its session once the bind commits off the event loop (#2850). + await host.historySettled(); + if (!host.isPrincipalAuthorized(principal.id)) return c.json({ error: "Authentication required", authRequired: true }, 401); const review = host.policyProposals?.read(principal.id, turnId, requestId, c.req.query("session")); if (!review) return c.json({ error: "No policy proposal receipt is available for this request." }, 404); return c.json(review); diff --git a/packages/ui-server/src/routes/session-profiles.ts b/packages/ui-server/src/routes/session-profiles.ts index 12809958f1..6a89284f73 100644 --- a/packages/ui-server/src/routes/session-profiles.ts +++ b/packages/ui-server/src/routes/session-profiles.ts @@ -28,7 +28,7 @@ export function createSessionProfileRoutes(profiles: SessionProfiles) { catch { throw new SessionProfileError(400, "Invalid JSON."); } if (!body || typeof body !== "object" || Array.isArray(body) || Object.keys(body).some(key => key !== "preference") || !["governed", "strict"].includes((body as { preference: string }).preference)) throw new SessionProfileError(400, "Choose governed or strict."); - return c.json(profiles.save(principal.id, kind, id, (body as { preference: "governed" | "strict" }).preference)); + return c.json(await profiles.save(principal.id, kind, id, (body as { preference: "governed" | "strict" }).preference)); } catch (error) { if (error instanceof SessionProfileError) return c.json({ error: error.message, ...(error.status === 401 ? { authRequired: true } : {}) }, error.status); throw error; diff --git a/packages/ui-server/src/routes/sessions.ts b/packages/ui-server/src/routes/sessions.ts index 48a81d38ed..5b0b8803e5 100644 --- a/packages/ui-server/src/routes/sessions.ts +++ b/packages/ui-server/src/routes/sessions.ts @@ -8,6 +8,11 @@ export function createSessionRoutes(deps: { registry: BackendRegistry; db: Database; prepareHistory?: (sessionId: string, messages: SessionHistoryMessage[]) => SessionHistoryMessage[]; + /** + * Settles once every catalog write submitted before it has settled: those + * commit off the event loop (#2850), and both routes read what they wrote. + */ + settled?: () => Promise; /** * Whether this host labels pills (#1004). Off, stored labels from an * earlier run are not listed: nothing would refresh them. @@ -51,6 +56,7 @@ export function createSessionRoutes(deps: { .get("/sessions", async (c) => { try { const backends = await registry.getBackends(); + await deps.settled?.(); // The catalog's accounting (total_cost_usd, num_turns) was write-only // for years on the default backend, whose listSessions hardcodes // zeros — merge the stored numbers in so the drawer's cost renders @@ -116,8 +122,10 @@ export function createSessionRoutes(deps: { .get("/sessions/:id", async (c) => { const id = c.req.param("id"); try { + await deps.settled?.(); const backend = await registry.getBackendForSession(getStoredBackendId(id)); const messages = await backend.getHistory(id); + await deps.settled?.(); return c.json({ id, messages: deps.prepareHistory?.(id, messages) ?? messages }); } catch (err) { return c.json( diff --git a/packages/ui-server/src/ws/bridge.ts b/packages/ui-server/src/ws/bridge.ts index bb91e5d1a2..e7301726f9 100644 --- a/packages/ui-server/src/ws/bridge.ts +++ b/packages/ui-server/src/ws/bridge.ts @@ -244,15 +244,19 @@ export function makeBridge( catalog.bindGovernanceEvents?.(turnId, msg.sessionId); host.inboxTools?.shares?.bind(turnId, msg.sessionId); if (workerProfile && host.sessionProfiles && turn.turnId === turnId) { + // Both commit after the row above (#2850); the frame reads them as submitted. + const named = msg.sessionId; + const unrecorded = (error: unknown) => host.log.emit({ severityText: "WARN", body: "session worker profile not recorded", + attributes: { "session.id": named, error: error instanceof Error ? error.message : String(error) } }); if (!profileBound && workerProfile.draftId && workerProfile.preference) { - host.sessionProfiles.bind(turn.principalId, workerProfile.draftId, msg.sessionId); + host.sessionProfiles.bind(turn.principalId, workerProfile.draftId, msg.sessionId).catch(unrecorded); profileBound = true; } - host.sessionProfiles.record(msg.sessionId, workerProfile.origin, workerProfile.actualProfile); + host.sessionProfiles.record(msg.sessionId, workerProfile.origin, workerProfile.actualProfile).catch(unrecorded); msg = { ...msg, workerProfile: host.sessionProfiles.read(turn.principalId, "session", msg.sessionId) }; } - host.policyProposals?.bind(turnId, msg.sessionId); + host.policyProposals?.bind(turnId, msg.sessionId).catch(() => { /* Its proposals keep a null session: unbound, never misattributed. */ }); // The session's pill label follows its latest request (#1004), asked // for once its row exists, so the answer always has a row to land on. if (!labelAsked) { diff --git a/packages/ui-server/src/ws/dispatch.ts b/packages/ui-server/src/ws/dispatch.ts index d943f6e5f0..b140d213f0 100644 --- a/packages/ui-server/src/ws/dispatch.ts +++ b/packages/ui-server/src/ws/dispatch.ts @@ -210,6 +210,8 @@ export async function handleClientMessage( case "retry_status": { try { + // Retained retries and their invalidations commit off the event loop (#2850). + await catalog.settled?.(); host.sendMessage(ws, catalog.retryReceipt?.(msg.sessionId, msg.requestId, connection.authorization.principalId) ?? { type: "retry_receipt", sessionId: msg.sessionId, requestId: msg.requestId, state: "unknown" }); } catch { @@ -225,6 +227,11 @@ export async function handleClientMessage( catch { host.sendMessage(ws, receipt); } }; try { + // Read and reserve against every retained request and invalidation + // submitted before this frame (#2850). One submitted during the file + // check below came with an accepted message, which the admission + // check after it refuses. + await catalog.settled?.(); // Replay a receipt even while its turn runs. A repeated request id // cannot become a queued follow-up, including after reconnect. const prior = catalog.retryReceipt?.(msg.sessionId, msg.requestId, connection.authorization.principalId); diff --git a/packages/ui-server/src/ws/handoff.ts b/packages/ui-server/src/ws/handoff.ts index 85eecae25a..5f2c462759 100644 --- a/packages/ui-server/src/ws/handoff.ts +++ b/packages/ui-server/src/ws/handoff.ts @@ -134,6 +134,7 @@ export async function startHandoff( coordinator.handoffs.set(handoff.handoffId, entry); let handedOver = false; let recorded = false; + let settled = false; const release = () => { // Fail closed: a destination that exists but whose link could not be // stored keeps its key here, so a retry still finds it rather than @@ -159,7 +160,9 @@ export async function startHandoff( let sourceTurns: number | null = null; try { const sourceBackend = await host.registry.getBackendForSession(sourceBackendId); - sourceTurns = host.prepareHistory(handoff.sourceSessionId, await sourceBackend.getHistory(handoff.sourceSessionId)) + const history = await sourceBackend.getHistory(handoff.sourceSessionId); + await host.historySettled(); + sourceTurns = host.prepareHistory(handoff.sourceSessionId, history) .filter((message) => message.role === "user").length; } catch { /* the marker goes at the end */ } if (!connection.authorization.valid) return; @@ -178,16 +181,19 @@ export async function startHandoff( sourceSessionId: handoff.sourceSessionId, onNamed: (sessionId) => { entry.sessionId = sessionId; - recorded = catalog.recordHandoff!(sessionId, handoff.handoffId, handoff.sourceSessionId, sourceTurns); - if (!recorded) { + // The key stays held until the link commits (#2850): until then a + // retry finds this destination here rather than in the catalog. + void catalog.recordHandoff!(sessionId, handoff.handoffId, handoff.sourceSessionId, sourceTurns).then((stored) => { + recorded = stored; + if (stored) { if (settled) release(); return; } host.log.emit({ severityText: "WARN", body: "handoff link not recorded", attributes: { "session.id": sessionId, "handoff.id": handoff.handoffId }, }); - } + }); }, - onSettled: release, + onSettled: () => { settled = true; release(); }, }, }); } catch (err) { @@ -306,7 +312,9 @@ export async function prepareHandoff( } let history: SessionHistoryMessage[]; try { - history = host.prepareHistory(msg.sourceSessionId, await backend.getHistory(msg.sourceSessionId)); + const replayed = await backend.getHistory(msg.sourceSessionId); + await host.historySettled(); + history = host.prepareHistory(msg.sourceSessionId, replayed); } catch { return answer({ state: "failed", message: "Couldn't load this chat's history." }); } diff --git a/packages/ui-server/src/ws/host.ts b/packages/ui-server/src/ws/host.ts index 13956ccdb5..5c28da6fd1 100644 --- a/packages/ui-server/src/ws/host.ts +++ b/packages/ui-server/src/ws/host.ts @@ -557,6 +557,9 @@ export class WsHost { const messages = await backend.getHistory(sessionId); if (!authorized()) return "unauthorized"; await this.failureReplay.wait(sessionId); + // Catalog rows and proposal binds commit off the event loop (#2850): + // the history is joined onto what was submitted before it was read. + await this.historySettled(); if (!authorized()) return "unauthorized"; if (read.changed) return "changed"; publish(messages); @@ -591,6 +594,15 @@ export class WsHost { } } + /** + * Settles once every catalog write and proposal bind submitted before it + * has settled (#2850). Await it before `prepareHistory`. + * @internal Host wiring. + */ + async historySettled(): Promise { + await Promise.all([this.catalog.settled?.(), this.policyProposals?.settled()]); + } + /** * Replayed history as the host sends it: each observed turn's id on the * answer that ended it (#964), the context blocks of locally diff --git a/packages/ui-server/src/ws/message-sources.ts b/packages/ui-server/src/ws/message-sources.ts index 78ae242804..0507853f05 100644 --- a/packages/ui-server/src/ws/message-sources.ts +++ b/packages/ui-server/src/ws/message-sources.ts @@ -43,7 +43,8 @@ export function saveMessageSource( sessionId: string, text: string, source: MessageSource, - effort?: { thinkingLevel?: ThinkingLevel; turnId: string; files?: SharedFileMeta[] } + effort?: { thinkingLevel?: ThinkingLevel; turnId: string; files?: SharedFileMeta[] }, + at = Date.now() ): void { const hash = messageTextHash(text); db.transaction(() => { @@ -53,7 +54,7 @@ export function saveMessageSource( db.prepare( `INSERT INTO message_sources (session_id, text_hash, ordinal, source, created_at, thinking_level, turn_id, track_files) VALUES (?, ?, ?, ?, ?, ?, ?, ?)` - ).run(sessionId, hash, row.n, source, Date.now(), effort?.thinkingLevel ?? null, effort?.turnId ?? null, effort?.files?.length ? JSON.stringify(effort.files) : null); + ).run(sessionId, hash, row.n, source, at, effort?.thinkingLevel ?? null, effort?.turnId ?? null, effort?.files?.length ? JSON.stringify(effort.files) : null); }).immediate(); } diff --git a/packages/ui-server/src/ws/recovery.ts b/packages/ui-server/src/ws/recovery.ts index 84c3c250ef..a413bd87ec 100644 --- a/packages/ui-server/src/ws/recovery.ts +++ b/packages/ui-server/src/ws/recovery.ts @@ -26,11 +26,16 @@ export function supportsSessionRecovery(host: Pick): boolean return typeof host.catalog.acceptWork === "function" && typeof host.catalog.latestWork === "function"; } +/** Each coordinator's newest acceptance per session: only its outcome orders the session. */ +const latestAcceptance = new WeakMap>(); + /** * Record that the host accepted a request for a known session, and return * its revision. A failed write leaves the session unordered: its recovery * reads `unknown` until a later acceptance is recorded, because the * persisted latest request is now older than the work actually accepted. + * The write commits off the event loop (#2850); acceptances settle in the + * order they were made, so the last one to settle decides. */ export function acceptRequest( host: Pick, @@ -39,12 +44,20 @@ export function acceptRequest( backendId: string | null ): number | undefined { if (!host.catalog.acceptWork) return undefined; - const revision = host.catalog.acceptWork(sessionId, requestId ?? null, backendId); + const { unorderedSessions } = host.coordinator; + const latest = latestAcceptance.get(host.coordinator) ?? new Map(); + latestAcceptance.set(host.coordinator, latest); + const mine = (latest.get(sessionId) ?? 0) + 1; + latest.set(sessionId, mine); + const revision = host.catalog.acceptWork(sessionId, requestId ?? null, backendId, (stored) => { + if (latest.get(sessionId) !== mine) return; + latest.delete(sessionId); + if (stored) unorderedSessions.delete(sessionId); else unorderedSessions.add(sessionId); + }); if (revision === null) { - host.coordinator.unorderedSessions.add(sessionId); + unorderedSessions.add(sessionId); return undefined; } - host.coordinator.unorderedSessions.delete(sessionId); return revision; } diff --git a/packages/ui-server/src/ws/session-catalog-writes.ts b/packages/ui-server/src/ws/session-catalog-writes.ts new file mode 100644 index 0000000000..45884aadce --- /dev/null +++ b/packages/ui-server/src/ws/session-catalog-writes.ts @@ -0,0 +1,165 @@ +/** + * 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 + * 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. + */ +import type { Database } from "bun:sqlite"; +import type { BillingMode, 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"; +import { workerActivity, type ActivityWriteOptions } from "../activity/store-writes.js"; +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 * as work from "./session-work.js"; +import { saveTurnFailure, type RecordedTurnFailure } from "./turn-failures.js"; + +export const url = import.meta.url; + +export const UPSERT_SESSION_SQL = `INSERT INTO sessions (id, title, created_at, last_active_at, total_cost_usd, num_turns, provider_id, backend_id) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + last_active_at = excluded.last_active_at, + total_cost_usd = total_cost_usd + excluded.total_cost_usd, + num_turns = num_turns + excluded.num_turns, + provider_id = COALESCE(excluded.provider_id, provider_id), + backend_id = COALESCE(excluded.backend_id, backend_id)`; + +/** One `sessions` upsert, with its clock taken where it was submitted. */ +export interface SessionRow { + id: string; title: string; at: number; costUsd: number; turns: number; providerId: string | null; backendId: string | null; +} + +export interface ClaudeUsageContext { billingMode?: BillingMode; pricingRoute?: PricingRoute; principalId?: string } + +/** + * Where Claude usage accounting writes its title and unstreamed runs. + * `activity` crosses to the worker (the app's Activity writer identity and + * pricing table); `local` is an inline writer's own store and pricing. + */ +export interface SessionCatalogWriteOptions { + activity?: ActivityWriteOptions; + local?: () => { store: ActivityStore; pricing: RollupPricing }; +} + +/** + * Claude usage reconciliation (`SessionCatalog.accountClaudeUsage`), in one + * immediate transaction: the turn's claim, its title and unstreamed runs, the + * session's cost and the reconciled state. + */ +export function accountClaudeUsage(db: Database, store: ActivityStore, pricing: RollupPricing, sessionId: string, backendId: string, + turnId: string, evidence: ClaudeUsageEvidence, resumed: boolean, context?: ClaudeUsageContext): void { + db.transaction(() => { + if (db.query("SELECT 1 FROM claude_usage_turns WHERE turn_id=?").get(turnId)) return; + const row = db.query("SELECT state_json FROM claude_session_usage WHERE session_id=? AND backend_id=?") + .get(sessionId, backendId) as { state_json: string } | null; + const previous = row ? JSON.parse(row.state_json) as ClaudeUsageState : undefined; + const result = reconcileClaudeUsage(previous, evidence.completed, evidence.cumulative, resumed, evidence.cumulativeCompleted); + const allocation = allocateClaudeTitleUsage(result.auxiliary, evidence.title, previous?.titleReceiptIds); + result.state.titleReceiptIds = allocation.seenIds; + result.state.partial ||= allocation.partial; + for (const [index, receipt] of allocation.title.entries()) { + const cost = receipt.costUsd ?? priceUsage(receipt.usage.perModel!, pricing, context?.pricingRoute)?.costUsd; + const runId = `${turnId}:claude-title:${index}`, spanId = `${runId}:root`; + store.startSpan({ runId, spanId, sessionId, principalId: context?.principalId, kind: "turn", origin: "session", + name: "Claude session title", attrs: { "brain.backend.id": backendId } }); + store.endSpan(spanId, { outcome: "success", usage: { + inputTokens: receipt.usage.inputTokens, outputTokens: receipt.usage.outputTokens, + cacheReadTokens: receipt.usage.cacheReadTokens, cacheCreationTokens: receipt.usage.cacheCreationTokens, + ...(cost !== undefined ? { costUsd: cost } : {}), + }, attrs: { "brain.usage.source": receipt.source, "brain.usage.attribution": "title_helper", + "brain.usage.receipt_id": receipt.receiptId, "brain.usage.coverage": "observed", + "gen_ai.usage.per_model": receipt.usage.perModel!, + ...(context?.billingMode ? { "brain.billing_mode": context.billingMode } : {}), + ...(context?.pricingRoute ? { "brain.pricing_route": context.pricingRoute } : {}), + } }); + store.rollupRun(runId); + if (cost !== undefined) db.query("UPDATE sessions SET total_cost_usd=total_cost_usd+? WHERE id=?").run(cost, sessionId); + } + const auxiliaryCost = allocation.unattributed?.perModel ? priceUsage(allocation.unattributed.perModel, pricing, context?.pricingRoute)?.costUsd : undefined; + // An independent run keeps main conversation/cache metrics clean. + // Unavailable coverage remains visible as an unpriced run; this is + // unstreamed usage, not a claim that a model was a title helper. + if (allocation.unattributed || result.state.partial && !previous?.partial) { + const runId = `${turnId}:claude-unstreamed`, spanId = `${runId}:root`; + store.startSpan({ runId, spanId, sessionId, principalId: context?.principalId, kind: "turn", origin: "session", + name: "Unstreamed Claude usage", attrs: { "brain.backend.id": backendId } }); + store.endSpan(spanId, { outcome: "success", usage: allocation.unattributed ? { + inputTokens: allocation.unattributed.inputTokens, outputTokens: allocation.unattributed.outputTokens, + cacheReadTokens: allocation.unattributed.cacheReadTokens, cacheCreationTokens: allocation.unattributed.cacheCreationTokens, + ...(auxiliaryCost !== undefined ? { costUsd: auxiliaryCost } : {}), + } : undefined, attrs: { + "brain.usage.source": "claude_session_cumulative_remainder", + "brain.usage.attribution": "unattributed", + ...(evidence.title ? { "brain.usage.title_evidence": JSON.stringify(evidence.title) } : {}), + "brain.usage.coverage": result.state.partial ? "partial" : "observed", + ...(allocation.unattributed?.perModel ? { "gen_ai.usage.per_model": allocation.unattributed.perModel } : {}), + ...(allocation.unattributed && context?.billingMode ? { "brain.billing_mode": context.billingMode } : {}), + ...(context?.pricingRoute ? { "brain.pricing_route": context.pricingRoute } : {}), + ...(evidence.reportedCostUsd !== undefined ? { "brain.usage.reported_session_cost_usd": evidence.reportedCostUsd } : {}), + } }); + store.rollupRun(runId); + if (auxiliaryCost !== undefined) db.query("UPDATE sessions SET total_cost_usd=total_cost_usd+? WHERE id=?").run(auxiliaryCost, sessionId); + } + db.query("INSERT INTO claude_session_usage VALUES (?,?,?) ON CONFLICT(session_id,backend_id) DO UPDATE SET state_json=excluded.state_json") + .run(sessionId, backendId, JSON.stringify(result.state)); + db.query("INSERT INTO claude_usage_turns VALUES (?,?,?)").run(turnId, sessionId, backendId); + }).immediate(); +} + +/** Worker side: every procedure is one of the catalog's synchronous writers, run on the write connection. */ +export function setup(db: Database, options: SessionCatalogWriteOptions = {}): WriteProcedures { + // Built on first use, as the catalog's own accounting was. + let accounting: { store: ActivityStore; pricing: () => RollupPricing } | undefined; + const account = () => accounting ??= (() => { + if (options.local) { const { store, pricing } = options.local(); return { store, pricing: () => pricing }; } + let standalone: RollupPricing | undefined; + const fallback = () => standalone ??= createModelPricing(resolveStandalonePricingConfig()); + if (!options.activity) { const pricing = fallback(); return { store: createActivityStore(db, { pricing }), pricing: () => pricing }; } + const activity = workerActivity(db, options.activity); + return { store: activity.store, pricing: activity.pricing ?? fallback }; + })(); + return { + upsertSession(row: SessionRow) { + db.prepare(UPSERT_SESSION_SQL).run(row.id, row.title, row.at, row.at, row.costUsd, row.turns, row.providerId, row.backendId); + }, + claudeUsage(sessionId: string, backendId: string, turnId: string, evidence: ClaudeUsageEvidence, resumed: boolean, context?: ClaudeUsageContext) { + const { store, pricing } = account(); + accountClaudeUsage(db, store, pricing(), sessionId, backendId, turnId, evidence, resumed, context); + }, + stateRoom(sessionId: string, observation: StateRoomObservation) { + db.query("INSERT INTO session_state_room (session_id, observation) VALUES (?, ?) ON CONFLICT(session_id) DO UPDATE SET observation = excluded.observation") + .run(sessionId, JSON.stringify(observation)); + }, + turnFailure: (sessionId: string, backendId: string, record: RecordedTurnFailure) => saveTurnFailure(db, sessionId, backendId, record), + acceptWork: (sessionId: string, input: { revision: number; requestId: string | null; backendId: string | null; at: number }) => + work.acceptWork(db, sessionId, input), + dispatchWork: (sessionId: string, input: { revision: number; turnId: string; backendId: string; startedAt: number }) => + work.dispatchWork(db, sessionId, input), + turnBoundary: (sessionId: string, backendId: string, boundary: work.TurnBoundary) => work.saveTurnBoundary(db, sessionId, backendId, boundary), + messageSource: (sessionId: string, text: string, source: MessageSource, effort: { thinkingLevel?: ThinkingLevel; turnId: string; files?: SharedFileMeta[] } | undefined, at: number) => + saveMessageSource(db, sessionId, text, source, effort, at), + effectiveThinkingLevel: (sessionId: string, turnId: string, level: ThinkingLevel) => saveEffectiveThinkingLevel(db, sessionId, turnId, level), + // A stub that failed to persist leaves no row to update: that is not a + // recorded link, and saying so keeps the caller's key fail-closed. + handoff: (sessionId: string, handoffId: string, sourceSessionId: string, sourceTurns: number | null) => + db.query("UPDATE sessions SET handoff_from = ?, handoff_id = ?, handoff_from_turns = ? WHERE id = ? RETURNING id") + .get(sourceSessionId, handoffId, sourceTurns, sessionId) !== null, + label(sessionId: string, label: string, source: string) { + db.prepare("UPDATE sessions SET label = ?, label_source = ? WHERE id = ?").run(label, source, sessionId); + }, + 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 ed2e40f5c9..d24b80bde4 100644 --- a/packages/ui-server/src/ws/session-catalog.ts +++ b/packages/ui-server/src/ws/session-catalog.ts @@ -1,7 +1,8 @@ -import { createActivityStore, priceUsage, type AwaitableActivityStore, type RollupPricing } from "../activity/store.js"; +import { createActivityStore, priceUsage, type ActivityStore, type AwaitableActivityStore, type RollupPricing } from "../activity/store.js"; +import type { ActivityPricingConfig } from "../activity/store-writes.js"; 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 type { ClaudeUsageEvidence } from "./claude-usage-accounting.js"; import type { BillingMode, PricingRoute } from "@schlessera/brain-ui-sdk/protocol"; import type { Database } from "bun:sqlite"; import type { Logger } from "@opentelemetry/api-logs"; @@ -18,7 +19,7 @@ import type { } from "@schlessera/brain-ui-sdk/protocol"; import { stateRoomObservationSchema } from "@schlessera/brain-ui-sdk/schemas"; -import { attachMessageSources, saveMessageSource, saveEffectiveThinkingLevel } from "./message-sources.js"; +import { attachMessageSources } from "./message-sources.js"; import { loadLocalExchanges, saveLocalExchange, @@ -28,7 +29,9 @@ import { import * as retries from "./retry-requests.js"; import * as work from "./session-work.js"; -import { attachTurnFailures, saveTurnFailure, type RecordedTurnFailure } from "./turn-failures.js"; +import { attachTurnFailures, type RecordedTurnFailure } from "./turn-failures.js"; +import * as writes from "./session-catalog-writes.js"; +import { procedureCaller, type ProcedureCall, type SqliteWriter } from "../db/write-worker.js"; import type { GovernanceEventInput } from "../governance/events.js"; import { createGovernanceWrites, type GovernanceWrites } from "../governance/event-writes.js"; @@ -50,11 +53,17 @@ export interface SessionCatalog { /** Observed terminal metadata at its backend-normalized assistant position. */ recordTurnFailure?(sessionId: string, backendId: string, record: RecordedTurnFailure): void; attachTurnFailures?(sessionId: string, messages: SessionHistoryMessage[]): SessionHistoryMessage[]; + /** + * Settles once every catalog write submitted before it has settled + * (#2850): a reader that joins committed rows onto history awaits it. + */ + settled?(): Promise; /** * Record an accepted request and return its revision (#964), or null when - * it could not be stored. Never throws: acceptance must not fail a turn. + * none could be assigned. `settled` hears whether it was stored. Never + * throws: acceptance must not fail a turn. */ - acceptWork?(sessionId: string, requestId: string | null, backendId: string | null): number | null; + acceptWork?(sessionId: string, requestId: string | null, backendId: string | null, settled?: (stored: boolean) => void): number | null; /** Record the turn that runs the request accepted at `revision`. */ dispatchWork?(sessionId: string, revision: number, turnId: string, backendId: string, startedAt: number): void; /** The latest accepted request, or null. Throws when the read fails. */ @@ -119,10 +128,10 @@ export interface SessionCatalog { loadLocalExchanges?(sessionId: string): LocalExchangeRecord[]; /** * Link a handoff destination to its source and its client-minted key - * (#61). Called once `session_info` names the destination. Returns false + * (#61). Called once `session_info` names the destination. Settles false * when it could not be stored. */ - recordHandoff?(sessionId: string, handoffId: string, sourceSessionId: string, sourceTurns: number | null): boolean; + recordHandoff?(sessionId: string, handoffId: string, sourceSessionId: string, sourceTurns: number | null): Promise; /** The destination already created for a handoff key, or null. */ findHandoff?(handoffId: string): string | null; /** @@ -152,18 +161,11 @@ export interface SessionCatalog { recordGovernanceEvent?(event: GovernanceEventInput): void; } -const UPSERT_SESSION_SQL = `INSERT INTO sessions (id, title, created_at, last_active_at, total_cost_usd, num_turns, provider_id, backend_id) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(id) DO UPDATE SET - last_active_at = excluded.last_active_at, - total_cost_usd = total_cost_usd + excluded.total_cost_usd, - num_turns = num_turns + excluded.num_turns, - provider_id = COALESCE(excluded.provider_id, provider_id), - backend_id = COALESCE(excluded.backend_id, backend_id)`; - /** SQLite-backed catalog over the app's own database (injected by createApp). */ export function createSessionCatalog(db: () => Database, log?: Logger, scratchPins?: { record(sessionId: string, backendId: string, messages: SessionHistoryMessage[]): void }, - accounting?: { pricing: RollupPricing; store: AwaitableActivityStore; governance?: Pick }): SessionCatalog { + accounting?: { pricing: RollupPricing; store: AwaitableActivityStore; governance?: Pick; + /** Where catalog rows commit: the app's write worker (#2850). Absent, they commit on `db` at once. */ + writer?: SqliteWriter; pricingConfig?: ActivityPricingConfig }): SessionCatalog { let accountingDeps = accounting; // Governance rows commit on the write worker when the app has one (#2850). let governanceWrites = accounting?.governance; @@ -185,76 +187,89 @@ export function createSessionCatalog(db: () => Database, log?: Logger, scratchPi }, }); }; + + // Every catalog write is one procedure call, committed in submission order + // (#2850). `tail` is the latest one's settlement: the worker settles calls + // in order, so once it has settled, so has every write submitted before it. + let caller: ProcedureCall | undefined; + let tail: Promise = Promise.resolve(); + const settle = (done: Promise) => { tail = done.then(() => {}, () => {}); }; + const submit = (name: string, ...args: unknown[]): Promise => { + let done: Promise; + try { + caller ??= procedureCaller(db(), { url: writes.url, setup: writes.setup }, accounting?.writer + ? { activity: { writer: accounting.store.writer, ...(accounting.pricingConfig ? { pricing: accounting.pricingConfig } : {}) } } + : { local: () => ({ store: getAccounting().store as ActivityStore, pricing: getAccounting().pricing }) }, accounting?.writer); + done = caller(name, ...args); + } catch (error) { done = Promise.reject(error); } + settle(done); + return done; + }; + + // The effects of writes submitted and not yet settled, per session, in + // submission order. A synchronous read on the turn path answers from the + // committed row with these applied, so a write is visible from its + // submission, as it was when it committed inline. Each effect goes when its + // 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 pendingWork = new Map work.AcceptedWork | null>>(); + const track = (effects: Map, sessionId: string, effect: E, done: Promise): Promise => { + const list = effects.get(sessionId) ?? []; + list.push(effect); + effects.set(sessionId, list); + void done.then(() => {}, () => {}).then(() => { + list.splice(list.indexOf(effect), 1); + if (list.length === 0 && effects.get(sessionId) === list) effects.delete(sessionId); + }); + return done; + }; + const committedRow = (sessionId: string) => db() + .query("SELECT provider_id AS providerId, backend_id AS backendId FROM sessions WHERE id = ?") + .get(sessionId) as { providerId: string | null; backendId: string | null } | null; + /** The latest non-null value a pending upsert gives (its COALESCE), else none. */ + const pendingValue = (sessionId: string, key: "providerId" | "backendId"): string | null => { + const effects = rowEffects.get(sessionId) ?? []; + for (let index = effects.length - 1; index >= 0; index--) { + const effect = effects[index]!; + if ("upsert" in effect && effect.upsert[key] !== null) return effect.upsert[key]; + } + return null; + }; + const pendingRow = (sessionId: string) => (rowEffects.get(sessionId) ?? []).some(effect => "upsert" in effect); + const storedBackendId = (sessionId: string) => pendingValue(sessionId, "backendId") ?? committedRow(sessionId)?.backendId ?? null; + const hasRow = (sessionId: string) => pendingRow(sessionId) || db().query("SELECT 1 FROM sessions WHERE id = ?").get(sessionId) !== null; + const upsert = (sessionId: string, promptText: string, providerId: string | null, backendId: string, costUsd: number, turns: number) => { + const row: writes.SessionRow = { id: sessionId, title: promptText.slice(0, 100), at: Date.now(), costUsd, turns, providerId, backendId }; + return track(rowEffects, sessionId, { upsert: { providerId, backendId } }, submit("upsertSession", row)); + }; + /** Highest revision issued per session; it never goes down, so a failed acceptance's revision is never reused. */ + const issued = new Map(); + const latestWork = (sessionId: string): work.AcceptedWork | null => { + let row = work.latestWork(db(), sessionId); + for (const effect of pendingWork.get(sessionId) ?? []) row = effect(row); + return row; + }; + // Retained retries stay on this connection, and so stay fail-closed: a + // failed invalidation still refuses the input that needed it. A failure + // retained before its session's row has committed is stored once that row + // has, unless an input has invalidated it meanwhile. + const retryGenerations = new Map(); return { + async settled() { let waiting: Promise; do { waiting = tail; await waiting; } while (waiting !== tail); }, accountClaudeUsage(sessionId, backendId, turnId, evidence, resumed, context) { let mainCost: number | undefined; try { - const { pricing, store } = getAccounting(); + const { pricing } = getAccounting(); mainCost = evidence.completed?.perModel ? priceUsage(evidence.completed.perModel, pricing, context?.pricingRoute)?.costUsd : undefined; - db().transaction(() => { - if (db().query("SELECT 1 FROM claude_usage_turns WHERE turn_id=?").get(turnId)) return; - const row = db().query("SELECT state_json FROM claude_session_usage WHERE session_id=? AND backend_id=?") - .get(sessionId, backendId) as { state_json: string } | null; - const previous = row ? JSON.parse(row.state_json) as ClaudeUsageState : undefined; - const result = reconcileClaudeUsage(previous, evidence.completed, evidence.cumulative, resumed, evidence.cumulativeCompleted); - const allocation = allocateClaudeTitleUsage(result.auxiliary, evidence.title, previous?.titleReceiptIds); - result.state.titleReceiptIds = allocation.seenIds; - result.state.partial ||= allocation.partial; - for (const [index, receipt] of allocation.title.entries()) { - const cost = receipt.costUsd ?? priceUsage(receipt.usage.perModel!, pricing, context?.pricingRoute)?.costUsd; - const runId = `${turnId}:claude-title:${index}`, spanId = `${runId}:root`; - store.startSpan({ runId, spanId, sessionId, principalId: context?.principalId, kind: "turn", origin: "session", - name: "Claude session title", attrs: { "brain.backend.id": backendId } }); - store.endSpan(spanId, { outcome: "success", usage: { - inputTokens: receipt.usage.inputTokens, outputTokens: receipt.usage.outputTokens, - cacheReadTokens: receipt.usage.cacheReadTokens, cacheCreationTokens: receipt.usage.cacheCreationTokens, - ...(cost !== undefined ? { costUsd: cost } : {}), - }, attrs: { "brain.usage.source": receipt.source, "brain.usage.attribution": "title_helper", - "brain.usage.receipt_id": receipt.receiptId, "brain.usage.coverage": "observed", - "gen_ai.usage.per_model": receipt.usage.perModel!, - ...(context?.billingMode ? { "brain.billing_mode": context.billingMode } : {}), - ...(context?.pricingRoute ? { "brain.pricing_route": context.pricingRoute } : {}), - } }); - store.rollupRun(runId); - if (cost !== undefined) db().query("UPDATE sessions SET total_cost_usd=total_cost_usd+? WHERE id=?").run(cost, sessionId); - } - const auxiliaryCost = allocation.unattributed?.perModel ? priceUsage(allocation.unattributed.perModel, pricing, context?.pricingRoute)?.costUsd : undefined; - // An independent run keeps main conversation/cache metrics clean. - // Unavailable coverage remains visible as an unpriced run; this is - // unstreamed usage, not a claim that a model was a title helper. - if (allocation.unattributed || result.state.partial && !previous?.partial) { - const runId = `${turnId}:claude-unstreamed`, spanId = `${runId}:root`; - store.startSpan({ runId, spanId, sessionId, principalId: context?.principalId, kind: "turn", origin: "session", - name: "Unstreamed Claude usage", attrs: { "brain.backend.id": backendId } }); - store.endSpan(spanId, { outcome: "success", usage: allocation.unattributed ? { - inputTokens: allocation.unattributed.inputTokens, outputTokens: allocation.unattributed.outputTokens, - cacheReadTokens: allocation.unattributed.cacheReadTokens, cacheCreationTokens: allocation.unattributed.cacheCreationTokens, - ...(auxiliaryCost !== undefined ? { costUsd: auxiliaryCost } : {}), - } : undefined, attrs: { - "brain.usage.source": "claude_session_cumulative_remainder", - "brain.usage.attribution": "unattributed", - ...(evidence.title ? { "brain.usage.title_evidence": JSON.stringify(evidence.title) } : {}), - "brain.usage.coverage": result.state.partial ? "partial" : "observed", - ...(allocation.unattributed?.perModel ? { "gen_ai.usage.per_model": allocation.unattributed.perModel } : {}), - ...(allocation.unattributed && context?.billingMode ? { "brain.billing_mode": context.billingMode } : {}), - ...(context?.pricingRoute ? { "brain.pricing_route": context.pricingRoute } : {}), - ...(evidence.reportedCostUsd !== undefined ? { "brain.usage.reported_session_cost_usd": evidence.reportedCostUsd } : {}), - } }); - store.rollupRun(runId); - if (auxiliaryCost !== undefined) db().query("UPDATE sessions SET total_cost_usd=total_cost_usd+? WHERE id=?").run(auxiliaryCost, sessionId); - } - db().query("INSERT INTO claude_session_usage VALUES (?,?,?) ON CONFLICT(session_id,backend_id) DO UPDATE SET state_json=excluded.state_json") - .run(sessionId, backendId, JSON.stringify(result.state)); - db().query("INSERT INTO claude_usage_turns VALUES (?,?,?)").run(turnId, sessionId, backendId); - }).immediate(); + submit("claudeUsage", sessionId, backendId, turnId, evidence, resumed, context).catch(error => reportWriteFailure(sessionId, error)); } catch (error) { reportWriteFailure(sessionId, error); } return mainCost; }, recordStateRoom(sessionId, observation) { try { const parsed = stateRoomObservationSchema.parse(observation); - db().query("INSERT INTO session_state_room (session_id, observation) VALUES (?, ?) ON CONFLICT(session_id) DO UPDATE SET observation = excluded.observation") - .run(sessionId, JSON.stringify(parsed)); + submit("stateRoom", sessionId, parsed).catch(error => reportWriteFailure(sessionId, error)); } catch (error) { reportWriteFailure(sessionId, error); } }, readStateRoom(sessionId) { @@ -268,25 +283,37 @@ export function createSessionCatalog(db: () => Database, log?: Logger, scratchPi catch (err) { reportWriteFailure(sessionId, err); } }, recordTurnFailure(sessionId, backendId, record) { - try { saveTurnFailure(db(), sessionId, backendId, record); } - catch (err) { reportWriteFailure(sessionId, err); } + submit("turnFailure", sessionId, backendId, record).catch(err => reportWriteFailure(sessionId, err)); }, attachTurnFailures(sessionId, messages) { try { - const row = db().query("SELECT backend_id FROM sessions WHERE id = ?").get(sessionId) as { backend_id: string | null } | null; - return row?.backend_id ? attachTurnFailures(db(), sessionId, row.backend_id, messages) : messages; + const backendId = storedBackendId(sessionId); + return backendId ? attachTurnFailures(db(), sessionId, backendId, messages) : messages; } catch (err) { reportWriteFailure(sessionId, err); return messages; } }, saveRetryRequest(sessionId, turnId, principalId, request, prompt, failure) { - try { retries.saveRetryRequest(db(), sessionId, turnId, principalId, request, prompt, failure); return true; } - catch (err) { reportWriteFailure(sessionId, err); return false; } + const save = () => retries.saveRetryRequest(db(), sessionId, turnId, principalId, request, prompt, failure); + try { + if (!pendingRow(sessionId)) { save(); return true; } + // Its row references the session's, still pending: it is stored once + // that has committed, and the answer advertises Retry now, since a + // frame cannot wait. If it is not kept, reserving it refuses. + const generation = retryGenerations.get(sessionId) ?? 0; + const stored = tail.then(() => { if ((retryGenerations.get(sessionId) ?? 0) === generation) save(); }); + settle(stored); + stored.catch(err => reportWriteFailure(sessionId, err)); + return true; + } catch (err) { reportWriteFailure(sessionId, err); return false; } }, // Retry reads/consumption must fail closed: unlike accounting, an // unsuccessful invalidation cannot permit an obsolete prompt to run. - clearRetryRequest(sessionId) { retries.clearRetryRequest(db(), sessionId); }, + clearRetryRequest(sessionId) { + retryGenerations.set(sessionId, (retryGenerations.get(sessionId) ?? 0) + 1); + retries.clearRetryRequest(db(), sessionId); + }, peekRetry(sessionId) { return retries.retained(db(), sessionId); }, reserveRetry(sessionId, turnId, requestId, principalId, expectedPrompt) { return retries.reserveRetry(db(), sessionId, turnId, requestId, principalId, expectedPrompt); }, refuseRetry(sessionId, requestId, principalId, message) { return retries.refuseRetry(db(), sessionId, requestId, principalId, message); }, @@ -295,26 +322,37 @@ export function createSessionCatalog(db: () => Database, log?: Logger, scratchPi try { return retries.attachRetryRequest(db(), sessionId, messages); } catch (err) { reportWriteFailure(sessionId, err); return messages; } }, - acceptWork(sessionId, requestId, backendId) { - try { return work.acceptWork(db(), sessionId, { requestId, backendId, at: Date.now() }); } - catch (err) { reportWriteFailure(sessionId, err); return null; } + acceptWork(sessionId, requestId, backendId, settled) { + let revision: number; + let latest: work.AcceptedWork | null; + const at = Date.now(); + try { + latest = latestWork(sessionId); + revision = Math.max(issued.get(sessionId) ?? 0, latest?.revision ?? 0) + 1; + } catch (err) { reportWriteFailure(sessionId, err); return null; } + issued.set(sessionId, revision); + // The statement's own effect: the upsert keeps the stored backend when none is given. + track(pendingWork, sessionId, (row: work.AcceptedWork | null) => ( + { revision, requestId, turnId: null, backendId: backendId ?? row?.backendId ?? null, acceptedAt: at, startedAt: null }), + submit("acceptWork", sessionId, { revision, requestId, backendId, at })) + .then(() => settled?.(true), err => { reportWriteFailure(sessionId, err); settled?.(false); }); + return revision; }, dispatchWork(sessionId, revision, turnId, backendId, startedAt) { - try { work.dispatchWork(db(), sessionId, { revision, turnId, backendId, startedAt }); } - catch (err) { reportWriteFailure(sessionId, err); } - }, - latestWork(sessionId) { return work.latestWork(db(), sessionId); }, - hasSession(sessionId) { - return db().query("SELECT 1 FROM sessions WHERE id = ?").get(sessionId) !== null; + // The statement's own effect: only the latest acceptance, not yet dispatched, takes the turn. + track(pendingWork, sessionId, (row: work.AcceptedWork | null) => row && row.revision === revision && row.turnId === null + ? { ...row, turnId, startedAt, backendId: row.backendId ?? backendId } : row, + submit("dispatchWork", sessionId, { revision, turnId, backendId, startedAt })).catch(err => reportWriteFailure(sessionId, err)); }, + latestWork, + hasSession: hasRow, recordTurnBoundary(sessionId, backendId, boundary) { - try { work.saveTurnBoundary(db(), sessionId, backendId, boundary); } - catch (err) { reportWriteFailure(sessionId, err); } + submit("turnBoundary", sessionId, backendId, boundary).catch(err => reportWriteFailure(sessionId, err)); }, attachTurnBoundaries(sessionId, messages) { try { - const row = db().query("SELECT backend_id FROM sessions WHERE id = ?").get(sessionId) as { backend_id: string | null } | null; - return row?.backend_id ? work.attachTurnBoundaries(db(), sessionId, row.backend_id, messages) : messages; + const backendId = storedBackendId(sessionId); + return backendId ? work.attachTurnBoundaries(db(), sessionId, backendId, messages) : messages; } catch (err) { // Without the boundaries a replay carries no turn ids, which is // what it carried before they existed: unproven, never wrong. @@ -323,69 +361,27 @@ export function createSessionCatalog(db: () => Database, log?: Logger, scratchPi } }, getStoredProviderId(sessionId) { - const row = db() - .query("SELECT provider_id AS providerId FROM sessions WHERE id = ?") - .get(sessionId) as { providerId: string | null } | null; - return row?.providerId ?? null; + return pendingValue(sessionId, "providerId") ?? committedRow(sessionId)?.providerId ?? null; }, - getStoredBackendId(sessionId) { - const row = db() - .query("SELECT backend_id AS backendId FROM sessions WHERE id = ?") - .get(sessionId) as { backendId: string | null } | null; - return row?.backendId ?? null; - }, + getStoredBackendId: storedBackendId, persistSessionStub(sessionId, promptText, providerId, backendId) { - try { - db() - .prepare(UPSERT_SESSION_SQL) - .run( - sessionId, - promptText.slice(0, 100), - Date.now(), - Date.now(), - 0, - 0, - providerId, - backendId - ); - } catch (err) { - reportWriteFailure(sessionId, err); - } + upsert(sessionId, promptText, providerId, backendId, 0, 0).catch(err => reportWriteFailure(sessionId, err)); }, persistSession(msg, promptText, providerId, backendId) { - try { - db() - .prepare(UPSERT_SESSION_SQL) - .run( - msg.sessionId, - promptText.slice(0, 100), - Date.now(), - Date.now(), - // costUsd is optional (rev 2): absent = unknown, accounted as 0. - msg.costUsd ?? 0, - msg.numTurns, - providerId, - backendId - ); - } catch (err) { - reportWriteFailure(msg.sessionId, err); - } + // costUsd is optional (rev 2): absent = unknown, accounted as 0. + upsert(msg.sessionId, promptText, providerId, backendId, msg.costUsd ?? 0, msg.numTurns).catch(err => reportWriteFailure(msg.sessionId, err)); }, recordMessageSource(sessionId, text, source, effort) { - try { - saveMessageSource(db(), sessionId, text, source, effort); - } catch (err) { - reportWriteFailure(sessionId, err); - } + // Its clock is taken now; its ordinal where it commits, in order. + submit("messageSource", sessionId, text, source, effort, Date.now()).catch(err => reportWriteFailure(sessionId, err)); }, recordEffectiveThinkingLevel(sessionId, turnId, level) { - try { saveEffectiveThinkingLevel(db(), sessionId, turnId, level); } - catch (err) { reportWriteFailure(sessionId, err); } + submit("effectiveThinkingLevel", sessionId, turnId, level).catch(err => reportWriteFailure(sessionId, err)); }, recordLocalExchange(sessionId, exchange, delivered) { @@ -426,17 +422,8 @@ export function createSessionCatalog(db: () => Database, log?: Logger, scratchPi }, recordHandoff(sessionId, handoffId, sourceSessionId, sourceTurns) { - try { - // A stub that failed to persist leaves no row to update: that is not - // a recorded link, and saying so keeps the caller's key fail-closed. - const result = db() - .prepare("UPDATE sessions SET handoff_from = ?, handoff_id = ?, handoff_from_turns = ? WHERE id = ?") - .run(sourceSessionId, handoffId, sourceTurns, sessionId); - return result.changes === 1; - } catch (err) { - reportWriteFailure(sessionId, err); - return false; - } + return submit("handoff", sessionId, handoffId, sourceSessionId, sourceTurns) + .catch((err: unknown) => { reportWriteFailure(sessionId, err); return false; }); }, findHandoff(handoffId) { @@ -448,6 +435,9 @@ export function createSessionCatalog(db: () => Database, log?: Logger, scratchPi sessionLabel(sessionId) { try { + // A label lands only on a row that exists by the time it commits. + const pending = [...rowEffects.get(sessionId) ?? []].reverse().find(effect => "label" in effect); + if (pending && "label" in pending && hasRow(sessionId)) return pending.label; const row = db() .query("SELECT label, label_source AS source FROM sessions WHERE id = ?") .get(sessionId) as { label: string | null; source: string | null } | null; @@ -458,13 +448,8 @@ export function createSessionCatalog(db: () => Database, log?: Logger, scratchPi }, saveSessionLabel(sessionId, label, source) { - try { - db() - .prepare("UPDATE sessions SET label = ?, label_source = ? WHERE id = ?") - .run(label, source, sessionId); - } catch (err) { - reportWriteFailure(sessionId, err); - } + track(rowEffects, sessionId, { label: { label, source } }, submit("label", sessionId, label, source)) + .catch(err => reportWriteFailure(sessionId, err)); }, bindGovernanceEvents(turnId, sessionId) { @@ -491,13 +476,7 @@ export function createSessionCatalog(db: () => Database, log?: Logger, scratchPi addSessionCost(sessionId, costUsd) { if (!Number.isFinite(costUsd) || costUsd <= 0) return; - try { - db() - .prepare("UPDATE sessions SET total_cost_usd = COALESCE(total_cost_usd, 0) + ? WHERE id = ?") - .run(costUsd, sessionId); - } catch (err) { - reportWriteFailure(sessionId, err); - } + submit("addCost", sessionId, costUsd).catch(err => reportWriteFailure(sessionId, err)); }, attachMessageSources(sessionId, messages) { diff --git a/packages/ui-server/src/ws/session-profile-writes.ts b/packages/ui-server/src/ws/session-profile-writes.ts new file mode 100644 index 0000000000..dcf76089f6 --- /dev/null +++ b/packages/ui-server/src/ws/session-profile-writes.ts @@ -0,0 +1,55 @@ +/** + * Session worker-profile writes off the event loop (#2850). A session's + * 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. + */ +import type { Database } from "bun:sqlite"; + +import type { WriteProcedures } from "../db/write-worker.js"; +import { isUsablePrincipal, resolvePrincipal } from "../db/principals.js"; + +export const url = import.meta.url; + +type Profile = "governed" | "strict"; + +export class SessionProfileError extends Error { + constructor(readonly status: 400 | 401 | 404 | 409 | 507, message: string) { super(message); } +} + +export function authorizeProfile(db: Database, principalId: string, now: number) { + const principal = resolvePrincipal(db, principalId); + if (!principal || !isUsablePrincipal(principal, now)) throw new SessionProfileError(401, "Authentication required."); +} + +export function draftProfile(db: Database, principalId: string, id: string) { + return db.query("SELECT preference FROM draft_worker_profiles WHERE principal_id = ? AND draft_id = ?").get(principalId, id) as { preference: Profile } | null; +} + +/** A new session takes its draft's preference; the draft is consumed. */ +export function bindDraftProfile(db: Database, principalId: string, draftId: string, sessionId: string, now: number) { + db.transaction(() => { + authorizeProfile(db, principalId, now); + if (!db.query("SELECT 1 FROM sessions WHERE id = ?").get(sessionId)) throw new SessionProfileError(404, "This chat is not known to the host."); + const row = draftProfile(db, principalId, draftId); + if (!row) throw new SessionProfileError(409, "This draft preference is unavailable."); + // A save during the first turn is for its NEXT admission; the running profile stays frozen. + db.query("INSERT INTO session_worker_profiles (session_id, preference, updated_at, updated_by) VALUES (?, ?, ?, ?) ON CONFLICT(session_id) DO NOTHING") + .run(sessionId, row.preference, now, principalId); + db.query("DELETE FROM draft_worker_profiles WHERE principal_id = ? AND draft_id = ?").run(principalId, draftId); + }).immediate(); +} + +/** The profile a session's latest turn was admitted with. */ +export function recordSessionProfile(db: Database, sessionId: string, origin: "interactive" | "voice", actualProfile: Profile, now: number) { + db.query("INSERT INTO session_worker_profiles (session_id, preference, updated_at, updated_by, origin, actual_profile) VALUES (?, 'governed', ?, 'host', ?, ?) ON CONFLICT(session_id) DO UPDATE SET origin = excluded.origin, actual_profile = excluded.actual_profile") + .run(sessionId, now, origin, actualProfile); +} + +export function setup(db: Database): WriteProcedures { + return { + 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 3c2dc595fc..0d098af94d 100644 --- a/packages/ui-server/src/ws/session-profiles.ts +++ b/packages/ui-server/src/ws/session-profiles.ts @@ -2,23 +2,63 @@ 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 { isUsablePrincipal, resolvePrincipal } from "../db/principals.js"; 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"; +import { authorizeProfile, draftProfile, SessionProfileError } from "./session-profile-writes.js"; + +export { SessionProfileError }; type Profile = "governed" | "strict"; -export class SessionProfileError extends Error { - constructor(readonly status: 400 | 401 | 404 | 409 | 507, message: string) { super(message); } -} -export function createSessionProfiles(db: Database, turns: GovernedTurns, budgets?: import("../brain/turn-budgets.js").TurnBudgets, limits?: { hardTimeoutMs: () => number }) { - const authorize = (principalId: string) => { - const principal = resolvePrincipal(db, principalId); - if (!principal || !isUsablePrincipal(principal, Date.now())) throw new SessionProfileError(401, "Authentication required."); +type SessionRow = { preference: Profile; actualProfile: Profile | null; origin: "interactive" | "voice" }; +type Effect = { bind: Profile } | { record: { origin: "interactive" | "voice"; actualProfile: Profile } }; + +export function createSessionProfiles(db: Database, turns: GovernedTurns, budgets?: import("../brain/turn-budgets.js").TurnBudgets, limits?: { hardTimeoutMs: () => number }, + /** + * Where a session's bind and record commit: the app's write worker + * (#2850), after the catalog row they reference. `catalog` answers whether + * a session's row exists, its pending row included, and when the + * catalog's writes have settled. + */ + 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); + // 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. + const pending = new Map(); + let tail: Promise = Promise.resolve(); + const write = (sessionId: string, effect: Effect, name: string, ...args: unknown[]) => { + const effects = pending.get(sessionId) ?? []; + effects.push(effect); + pending.set(sessionId, effects); + const done = submit(name, ...args); + tail = done.then(() => {}, () => {}); + void tail.then(() => { + effects.splice(effects.indexOf(effect), 1); + if (effects.length === 0 && pending.get(sessionId) === effects) pending.delete(sessionId); + }); + return done; }; + const exists = (id: string) => options.catalog?.hasSession?.(id) ?? db.query("SELECT 1 FROM sessions WHERE id = ?").get(id) !== null; const session = (id: string) => { - if (!db.query("SELECT 1 FROM sessions WHERE id = ?").get(id)) throw new SessionProfileError(404, "This chat is not known to the host."); - return db.query("SELECT preference, actual_profile AS actualProfile, origin FROM session_worker_profiles WHERE session_id = ?").get(id) as { preference: Profile; actualProfile: Profile | null; origin: "interactive" | "voice" } | null; + if (!exists(id)) throw new SessionProfileError(404, "This chat is not known to the host."); + let row = db.query("SELECT preference, actual_profile AS actualProfile, origin FROM session_worker_profiles WHERE session_id = ?").get(id) as SessionRow | null; + for (const effect of pending.get(id) ?? []) { + if ("bind" in effect) row ??= { preference: effect.bind, actualProfile: null, origin: "interactive" }; + else row = row ? { ...row, ...effect.record } : { preference: "governed", ...effect.record }; + } + return row; + }; + // A bind consumes its draft: once submitted, the draft reads as gone. + const consumed = new Set(); + const draft = (principalId: string, id: string) => consumed.has(JSON.stringify([principalId, id])) ? null : draftProfile(db, principalId, id); + /** Settles once every bind and record submitted before it, and the catalog rows they need, have settled. */ + const settled = async () => { + await options.catalog?.settled?.(); + let waiting: Promise; + do { waiting = tail; await waiting; } while (waiting !== tail); }; - const draft = (principalId: string, id: string) => db.query("SELECT preference FROM draft_worker_profiles WHERE principal_id = ? AND draft_id = ?").get(principalId, id) as { preference: Profile } | null; const facts = (preference: Profile, origin: "interactive" | "voice", actualProfile?: Profile): SessionWorkerProfile => { const resolved = turns.facts(origin); const caps = scopeLimits(resolved.policy.policy); @@ -44,7 +84,10 @@ export function createSessionProfiles(db: Database, turns: GovernedTurns, budget }; return { read, - save(principalId: string, kind: "session" | "draft", id: string, preference: Profile) { + settled, + // A saved preference still commits here, after what it reads has settled. + 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."); @@ -71,20 +114,19 @@ export function createSessionProfiles(db: Database, turns: GovernedTurns, budget if (!row) throw new SessionProfileError(409, "This draft preference is unavailable. Refresh before sending."); return row.preference; }, - bind(principalId: string, draftId: string, sessionId: string) { - db.transaction(() => { - authorize(principalId); session(sessionId); - const row = draft(principalId, draftId); - if (!row) throw new SessionProfileError(409, "This draft preference is unavailable."); - // A save during the first turn is for its NEXT admission; the running profile stays frozen. - db.query("INSERT INTO session_worker_profiles (session_id, preference, updated_at, updated_by) VALUES (?, ?, ?, ?) ON CONFLICT(session_id) DO NOTHING") - .run(sessionId, row.preference, Date.now(), principalId); - db.query("DELETE FROM draft_worker_profiles WHERE principal_id = ? AND draft_id = ?").run(principalId, draftId); - }).immediate(); + /** Checked now, as the inline write did; it commits after the session's row. */ + bind(principalId: string, draftId: string, sessionId: string): Promise { + authorize(principalId); session(sessionId); + const row = draft(principalId, draftId); + if (!row) throw new SessionProfileError(409, "This draft preference is unavailable."); + const key = JSON.stringify([principalId, draftId]); + consumed.add(key); + const done = write(sessionId, { bind: row.preference }, "bind", principalId, draftId, sessionId, Date.now()); + void done.then(() => {}, () => {}).then(() => consumed.delete(key)); + return done; }, - record(sessionId: string, origin: "interactive" | "voice", actualProfile: Profile) { - db.query("INSERT INTO session_worker_profiles (session_id, preference, updated_at, updated_by, origin, actual_profile) VALUES (?, 'governed', ?, 'host', ?, ?) ON CONFLICT(session_id) DO UPDATE SET origin = excluded.origin, actual_profile = excluded.actual_profile") - .run(sessionId, Date.now(), origin, actualProfile); + record(sessionId: string, origin: "interactive" | "voice", actualProfile: Profile): Promise { + return write(sessionId, { record: { origin, actualProfile } }, "record", sessionId, origin, actualProfile, Date.now()); }, }; } diff --git a/packages/ui-server/src/ws/session-work.ts b/packages/ui-server/src/ws/session-work.ts index 00bda78a02..4c5bd9bec5 100644 --- a/packages/ui-server/src/ws/session-work.ts +++ b/packages/ui-server/src/ws/session-work.ts @@ -22,21 +22,22 @@ export interface AcceptedWork { } /** - * Record that the host accepted a request for `sessionId` and return its - * revision: one more than the last. The read and the write are one - * statement, so two acceptances can never share a revision. + * Record that the host accepted a request for `sessionId` at `revision`. The + * catalog assigns revisions where it accepts, one above the highest it has + * issued or seen committed (#2850), so two acceptances never share one and a + * failed write's revision is never issued again; a gap orders as before. */ export function acceptWork( db: Database, sessionId: string, - input: { requestId: string | null; backendId: string | null; at: number } + input: { revision: number; requestId: string | null; backendId: string | null; at: number } ): number { const row = db .query( `INSERT INTO session_work (session_id, revision, request_id, turn_id, backend_id, accepted_at, started_at) - VALUES (?, 1, ?, NULL, ?, ?, NULL) + VALUES (?, ?, ?, NULL, ?, ?, NULL) ON CONFLICT(session_id) DO UPDATE SET - revision = revision + 1, + revision = excluded.revision, request_id = excluded.request_id, turn_id = NULL, backend_id = COALESCE(excluded.backend_id, backend_id), @@ -44,7 +45,7 @@ export function acceptWork( started_at = NULL RETURNING revision` ) - .get(sessionId, input.requestId, input.backendId, input.at) as { revision: number }; + .get(sessionId, input.revision, input.requestId, input.backendId, input.at) as { revision: number }; return row.revision; } diff --git a/packages/ui-server/tests/app-writes-offloop.test.ts b/packages/ui-server/tests/app-writes-offloop.test.ts index b2bae87f66..27332d2ff5 100644 --- a/packages/ui-server/tests/app-writes-offloop.test.ts +++ b/packages/ui-server/tests/app-writes-offloop.test.ts @@ -16,6 +16,15 @@ import { createActivityStore } from "../src/activity/store.js"; import { createNativeJournalFixture } from "./helpers/native-journal-fixture.js"; 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 { createPolicyProposals } from "../src/brain/policy-proposal.js"; +import { createSessionRoutes } from "../src/routes/sessions.js"; +import { createStaticBackendRegistry } from "../src/agent/backend.js"; +import { acceptRequest } from "../src/ws/recovery.js"; +import { makeFakeBackend } from "./helpers/fake-backend.js"; +import { WsHost } from "../src/ws/host.js"; function fixture() { const root = mkdtempSync(join(tmpdir(), "app-writes-odysseus-")), brain = join(root, "brain"), path = join(root, "ui.db"); @@ -136,3 +145,155 @@ test("turn budget writes commit on the write worker at FULL, after earlier write expect(new Set(log.slice(1)), "BUDGET_TABLES_WRITTEN_ON_WORKER").toEqual(new Set(["turn_budget_usage", "turn_budget_days", "turn_budget_warnings", "provider_relay_usage"])); } finally { runtime.close(); await f.close(); } }); + +const CATALOG_TABLES = ["sessions", "session_work", "turn_boundaries", "message_sources", "turn_failures", "session_state_room", + "claude_session_usage", "claude_usage_turns", "session_worker_profiles", "draft_worker_profiles", "policy_proposals", "activity_spans"]; +const profileFacts = { facts: () => ({ policyProfile: "governed" as const, policy: { policy: resolveGovernancePolicy({ schema_version: 1 }), hash: null } }) } as never; + +test("session catalog rows, and the profile and proposal rows that reference them, commit on the write worker at FULL, in order", async () => { + const f = fixture(); + const runtime = createActivityRuntime(f.db, { log: quietLog, writer: f.writer }); + try { + const athena = createPrincipal(f.db, { authMethod: "password", label: "Athena device", ttlSeconds: 3600 }); + f.db.query("INSERT INTO draft_worker_profiles (principal_id, draft_id, preference, updated_at) VALUES (?, 'troy-draft', 'strict', 0)").run(athena.id); + f.db.query("INSERT INTO policy_proposals (request_id, turn_id, principal_id, session_id, input_hash, review, created_at) VALUES ('wooden-horse', 'voyage', ?, NULL, 'h', '{}', 0)").run(athena.id); + const sites = await watchWriteSites(f.db, f.writer, CATALOG_TABLES); + const pricing = { resolve: () => ({ input: 1, output: 1, cacheRead: 1, cacheWrite: 1, estimate: false, source: "snapshot" as const }) }; + const failures: string[] = []; + const reporting = { emit: (record: { attributes?: { error?: string } }) => { failures.push(String(record.attributes?.error)); } } as unknown as Logger; + const catalog = createSessionCatalog(() => f.db, reporting, undefined, { pricing, store: runtime.store, governance: runtime.governanceWrites, writer: f.writer }); + const profiles = createSessionProfiles(f.db, profileFacts, undefined, undefined, { writer: f.writer, catalog }); + const proposals = createPolicyProposals({ root: f.brain, db: f.db, writer: f.writer, lock: { withLock: (_key: string, run: () => unknown) => run() } as never, + observeUserPolicy: () => {}, recordEvent: () => {} }); + void runtime.store.startSpan({ spanId: "voyage:root", runId: "voyage", name: "invoke_agent", kind: "turn", origin: "session", sessionId: "ithaca" }); + // Everything below is submitted while an earlier commit holds the worker. + const held = f.writer.channel(probe, { table: "odysseus_log" }); await held.ready; + const slow = held.call("slow", "Calypso", 300); + // session_info: the stub, then the rows that reference it. + catalog.persistSessionStub("troy", "Sail for Troy", "claude-sonnet", "claude"); + void profiles.bind(athena.id, "troy-draft", "troy"); + void profiles.record("troy", "interactive", "governed"); + void proposals.bind("voyage", "troy"); + expect(f.db.query("SELECT 1 FROM sessions WHERE id = 'troy'").get(), "CATALOG_ROW_NOT_COMMITTED_ON_LOOP").toBeNull(); + expect([catalog.hasSession!("troy"), catalog.getStoredBackendId("troy"), catalog.getStoredProviderId("troy")], "CATALOG_PENDING_ROW_READS") + .toEqual([true, "claude", "claude-sonnet"]); + expect(profiles.read(athena.id, "session", "troy"), "PROFILE_PENDING_READ").toMatchObject({ preference: "strict", actualProfile: "governed" }); + expect(() => profiles.preference(athena.id, undefined, "troy-draft"), "PROFILE_PENDING_DRAFT_CONSUMED").toThrow("unavailable"); + expect([catalog.acceptWork!("troy", "req-1", "claude"), catalog.acceptWork!("troy", "req-2", null)], "WORK_REVISIONS_AT_ACCEPTANCE").toEqual([1, 2]); + catalog.dispatchWork!("troy", 2, "turn-2", "claude", 7); + expect(catalog.latestWork!("troy"), "WORK_PENDING_LATEST").toMatchObject({ revision: 2, requestId: "req-2", turnId: "turn-2", backendId: "claude", startedAt: 7 }); + catalog.recordMessageSource!("troy", "Sail for Troy", "voice-dictate", { turnId: "turn-2", thinkingLevel: "high" }); + catalog.recordEffectiveThinkingLevel!("troy", "turn-2", "medium"); + catalog.recordTurnBoundary!("troy", "claude", { turnId: "turn-2", assistantOrdinal: 0, prefixDigest: "d" }); + catalog.recordTurnFailure!("troy", "claude", { turnId: "turn-2", assistantOrdinal: 0, prefixDigest: "d", failure: { errorClass: "overloaded", message: "Poseidon" } as never }); + catalog.recordStateRoom!("troy", { usedBytes: 1, allowanceBytes: 10, observedAt: 1 } as never); + catalog.accountClaudeUsage!("troy", "claude", "turn-2", {}, false); + catalog.persistSession({ type: "result", sessionId: "troy", numTurns: 1, costUsd: 0.5 } as never, "Sail for Troy", null, "claude"); + catalog.saveSessionLabel!("troy", "Troy", "source-1"); + expect(catalog.sessionLabel!("troy"), "LABEL_PENDING_READ").toEqual({ label: "Troy", source: "source-1" }); + catalog.addSessionCost!("troy", 0.25); + const linked = catalog.recordHandoff!("troy", "h-troy-0001", "ithaca", 3); + await slow; + await catalog.settled!(); + expect(failures, "CATALOG_WRITES_COMMIT_ON_WORKER_AT_FULL").toEqual([]); + expect(await linked, "HANDOFF_LINK_COMMITS_ON_WORKER").toBe(true); + const log = await sites.log(); + expect(log[0], "CATALOG_AFTER_EARLIER_ACTIVITY_WRITE").toBe("activity_spans"); + const first = (table: string) => log.indexOf(table); + expect(first("sessions"), "CATALOG_STUB_BEFORE_ITS_REFERENCES").toBeLessThan(first("session_worker_profiles")); + expect(first("session_worker_profiles"), "CATALOG_PROFILE_BEFORE_PROPOSAL_BIND").toBeLessThan(first("policy_proposals")); + expect(new Set(log), "CATALOG_TABLES_WRITTEN_ON_WORKER").toEqual(new Set(CATALOG_TABLES)); + const row = f.db.query("SELECT backend_id, provider_id, total_cost_usd, num_turns, label, handoff_id FROM sessions WHERE id = 'troy'").get(); + expect(row, "CATALOG_ROW_COMMITTED").toEqual({ backend_id: "claude", provider_id: "claude-sonnet", total_cost_usd: 0.75, num_turns: 1, label: "Troy", handoff_id: "h-troy-0001" }); + expect(f.db.query("SELECT session_id FROM policy_proposals").get(), "PROPOSAL_BOUND_AFTER_STUB").toEqual({ session_id: "troy" }); + expect(catalog.latestWork!("troy"), "WORK_COMMITTED_LATEST").toMatchObject({ revision: 2, turnId: "turn-2" }); + } finally { runtime.close(); await f.close(); } +}); + +test("history readers wait for pending catalog writes; a failed acceptance's revision is never issued again", async () => { + const f = fixture(); + const runtime = createActivityRuntime(f.db, { log: quietLog, writer: f.writer }); + try { + const catalog = createSessionCatalog(() => f.db, quietLog, undefined, { pricing: { resolve: () => null }, store: runtime.store, writer: f.writer }); + const history = (id: string) => [{ role: "user" as const, content: `Sail for ${id}`, toolCalls: [] }]; + const registry = createStaticBackendRegistry([makeFakeBackend({ id: "claude", sessions: [] }), + makeFakeBackend({ id: "pi", sessions: [{ id: "troy", title: "Troy", createdAt: 1, lastActiveAt: 1, totalCostUsd: 0, numTurns: 0 }], histories: { troy: history("Troy"), "ithaca-return": history("Ithaca") } })], "claude"); + const routes = createSessionRoutes({ registry, db: f.db, settled: () => catalog.settled!() }); + const held = f.writer.channel(probe, { table: "odysseus_log" }); await held.ready; + const slow = held.call("slow", "Calypso", 300); + catalog.persistSessionStub("troy", "Sail for Troy", null, "pi"); + catalog.persistSession({ type: "result", sessionId: "troy", numTurns: 2, costUsd: 0.5 } as never, "Sail for Troy", null, "pi"); + const listed = await (await routes.request("/sessions")).json() as { sessions: Array<{ id: string; totalCostUsd: number }> }; + expect(listed.sessions.find(session => session.id === "troy")?.totalCostUsd, "SESSION_LIST_WAITS_FOR_CATALOG").toBe(0.5); + await slow; + // Its own held window: the history route picks the backend from the row. + const again = held.call("slow", "Calypso", 300); + catalog.persistSessionStub("ithaca-return", "Sail home", null, "pi"); + const replay = await (await routes.request("/sessions/ithaca-return")).json() as { messages: unknown[] }; + expect(replay.messages, "SESSION_HISTORY_WAITS_FOR_ITS_BACKEND_ROW").toHaveLength(1); + await again; + } finally { runtime.close(); await f.close(); } + const db = createUiDb(":memory:"); + try { + db.exec("CREATE TRIGGER refuse_odysseus BEFORE INSERT ON session_work WHEN NEW.request_id = 'refused' BEGIN SELECT RAISE(ABORT, 'Scylla'); END"); + const catalog = createSessionCatalog(() => db, quietLog); + const host = { catalog, coordinator: { unorderedSessions: new Set() } } as never as Parameters[0]; + expect(acceptRequest(host, "troy", "refused", null), "WORK_FAILED_ACCEPTANCE_HAS_A_REVISION").toBe(1); + await catalog.settled!(); + expect(host.coordinator.unorderedSessions.has("troy"), "WORK_FAILED_ACCEPTANCE_UNORDERS").toBe(true); + expect(acceptRequest(host, "troy", "req-2", null), "WORK_REVISION_NOT_REISSUED").toBe(2); + await catalog.settled!(); + expect(host.coordinator.unorderedSessions.has("troy"), "WORK_LATER_ACCEPTANCE_ORDERS").toBe(false); + } finally { db.close(); } +}); + +test("a failure retained before its session's row commits is stored after it, unless an input invalidated it", 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 failure = { errorClass: "overloaded", message: "Poseidon" } as never; + const request = { type: "chat_message", text: "Sail for Troy", attachments: [] } as never; + const held = f.writer.channel(probe, { table: "odysseus_log" }); await held.ready; + let slow = held.call("slow", "Calypso", 200); + catalog.persistSessionStub("troy", "Sail for Troy", null, "claude"); + expect(catalog.saveRetryRequest!("troy", "turn-1", "athena", request, "Sail for Troy", failure), "RETRY_ADVERTISED_WHILE_ROW_PENDING").toBe(true); + await slow; await catalog.settled!(); + expect(catalog.peekRetry!("troy")?.turnId, "RETRY_STORED_AFTER_ITS_ROW").toBe("turn-1"); + slow = held.call("slow", "Calypso", 200); + catalog.persistSessionStub("troy", "Sail for Troy", null, "claude"); + catalog.saveRetryRequest!("troy", "turn-2", "athena", request, "Sail for Troy", failure); + catalog.clearRetryRequest!("troy"); + await slow; await catalog.settled!(); + expect(catalog.peekRetry!("troy"), "RETRY_INVALIDATED_BEFORE_IT_WAS_STORED").toBeNull(); + } finally { await f.close(); } +}); + +test("a replayed history joins the catalog rows submitted before it was read", async () => { + const f = fixture(); + const runtime = createActivityRuntime(f.db, { log: quietLog, writer: f.writer }); + try { + const catalog = createSessionCatalog(() => f.db, quietLog, undefined, { pricing: { resolve: () => null }, store: runtime.store, writer: f.writer }); + const backend = makeFakeBackend({ id: "claude", histories: { troy: [{ role: "user", content: "Sail for Troy", toolCalls: [] }] } }); + const host = new WsHost({ registry: createStaticBackendRegistry([backend], "claude"), catalog, maxConcurrentSessions: () => 3 }); + const held = f.writer.channel(probe, { table: "odysseus_log" }); await held.ready; + const slow = held.call("slow", "Calypso", 300); + catalog.persistSessionStub("troy", "Sail for Troy", null, "claude"); + catalog.recordMessageSource!("troy", "Sail for Troy", "voice-dictate"); + let replayed: Array<{ source?: string }> = []; + await host.publishHistory("troy", backend, () => true, messages => { replayed = host.prepareHistory("troy", messages); }); + expect(replayed.map(message => message.source), "HISTORY_JOINS_PENDING_CATALOG_WRITES").toEqual(["voice-dictate"]); + await slow; + } finally { runtime.close(); await f.close(); } +}); + +test("only a session's newest acceptance orders or unorders it", () => { + const settle: Array<(stored: boolean) => void> = []; + let next: number | null = 1; + const host = { catalog: { acceptWork: (_s: string, _r: string | null, _b: string | null, settled?: (stored: boolean) => void) => { settle.push(settled!); return next; } }, + coordinator: { unorderedSessions: new Set() } } as never as Parameters[0]; + expect(acceptRequest(host, "troy", "req-1", null)).toBe(1); + next = null; + expect(acceptRequest(host, "troy", "req-2", null), "WORK_UNASSIGNED_ACCEPTANCE").toBeUndefined(); + settle[0]!(true); + expect(host.coordinator.unorderedSessions.has("troy"), "WORK_OLDER_ACCEPTANCE_DOES_NOT_REORDER").toBe(true); +}); diff --git a/packages/ui-server/tests/session-routes-handoff.test.ts b/packages/ui-server/tests/session-routes-handoff.test.ts index 97346b4eb2..8de67a158e 100644 --- a/packages/ui-server/tests/session-routes-handoff.test.ts +++ b/packages/ui-server/tests/session-routes-handoff.test.ts @@ -13,7 +13,7 @@ describe("GET /sessions handoff links (#61)", () => { const catalog = createSessionCatalog(() => db); catalog.persistSessionStub("src", "Plan the return", "claude", "claude"); catalog.persistSessionStub("dst", "Odysseus is sailing home", "pi", "pi"); - expect(catalog.recordHandoff!("dst", "h-ithaca-0001", "src", 4)).toBe(true); + expect(await catalog.recordHandoff!("dst", "h-ithaca-0001", "src", 4)).toBe(true); const registry = createStaticBackendRegistry([ makeFakeBackend({ id: "claude", sessions: [listed("src", "Ithaca return")] }), makeFakeBackend({ id: "pi", sessions: [listed("dst", "Odysseus is sailing home")] }), @@ -26,9 +26,9 @@ describe("GET /sessions handoff links (#61)", () => { expect(catalog.findHandoff!("h-ithaca-0001")).toBe("dst"); // The key is unique: a second destination cannot claim it. catalog.persistSessionStub("dst2", "again", "pi", "pi"); - expect(catalog.recordHandoff!("dst2", "h-ithaca-0001", "src", 4)).toBe(false); + expect(await catalog.recordHandoff!("dst2", "h-ithaca-0001", "src", 4)).toBe(false); // A destination whose stub never persisted has no row: not recorded. - expect(catalog.recordHandoff!("never-stored", "h-ithaca-0002", "src", 1)).toBe(false); + expect(await catalog.recordHandoff!("never-stored", "h-ithaca-0002", "src", 1)).toBe(false); expect(catalog.findHandoff!("h-ithaca-0002")).toBeNull(); db.close(); }); diff --git a/packages/ui-server/tests/ws-handoff.test.ts b/packages/ui-server/tests/ws-handoff.test.ts index 60f0698912..ff20b9281b 100644 --- a/packages/ui-server/tests/ws-handoff.test.ts +++ b/packages/ui-server/tests/ws-handoff.test.ts @@ -622,7 +622,7 @@ describe("review findings: authority, admission and fail-closed idempotency", () }); for (const pair of PAIRS) test(`${pair.source} → ${pair.destination}: a destination whose link could not be stored is still found by its key`, async () => { - const r = rig({ recordHandoff: () => () => false }); + const r = rig({ recordHandoff: () => async () => false }); const source = seedSource(r, pair.source); const ws = r.open(); const frame = { type: "chat_message", text: "Odysseus is sailing home.", providerId: pair.profile, requestId: "req-unrecorded", diff --git a/packages/ui-server/tests/ws-scope-caps.test.ts b/packages/ui-server/tests/ws-scope-caps.test.ts index 0f3e5a4265..98b789f502 100644 --- a/packages/ui-server/tests/ws-scope-caps.test.ts +++ b/packages/ui-server/tests/ws-scope-caps.test.ts @@ -53,7 +53,10 @@ for (const { adapter, voice, overshoot } of cases) test(`${adapter}: ${overshoot expect(heartbeatSize(), "SCOPE_ADAPTER_HEARTBEAT_GROWS_WHILE_RUNNING").toBeGreaterThan(1); yield { type: "system", subtype: "init", session_id: "11111111-1111-4111-8111-111111111111" }; for (let call = 1; call <= 4; call++) { - while (call > 2 && params.options.abortController.signal.aborted === false && paused()) await Bun.sleep(10); + // A frozen worker makes no call: this in-process stream stands in + // for one, so it holds while the host has the turn paused, not only + // once the pause summary reaches the client after its commit (#2850). + while (call > 2 && params.options.abortController.signal.aborted === false && (paused() || hostPaused())) await Bun.sleep(10); if (params.options.abortController.signal.aborted) break; yield { type: "assistant", message: { content: [{ type: "tool_use", id: `loom-${call}`, name: "Bash", input: { command: "printf Ithaca" } }] } }; yield { type: "user", message: { content: [{ type: "tool_result", tool_use_id: `loom-${call}`, content: "Ithaca", is_error: false }] } }; @@ -68,6 +71,7 @@ for (const { adapter, voice, overshoot } of cases) test(`${adapter}: ${overshoot const frames: ServerMessage[] = []; const latest = () => frames.filter((frame): frame is Extract => frame.type === "turn_journal").at(-1); const paused = () => latest()?.journal.scope?.receipts.at(-1)?.state === "paused"; + const hostPaused = () => [...app?.wsHost.coordinator.running ?? []].some(turn => app!.wsHost.scope?.isPaused(turn.turnId)); let releaseObservation!: () => void, observationHeld = false; const observationReleased = new Promise(done => { releaseObservation = done; }); let captureSpy: ReturnType | undefined; From 37e64a71981fb0d04b7c537b1542b9bdc0a6bbb9 Mon Sep 17 00:00:00 2001 From: Alain Schlesser Date: Sun, 25 Oct 2026 18:41:08 +0100 Subject: [PATCH 2/4] test(ui-server): wait for a failed catalog write before reading its warning Catalog writes settle off the event loop (#2850), so a failed write's warning is reported once its promise settles. Refs #2850 Co-Authored-By: Claude Opus 5.5 --- packages/ui-server/tests/ws-observability.test.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/packages/ui-server/tests/ws-observability.test.ts b/packages/ui-server/tests/ws-observability.test.ts index 6fa359b439..b10a8b68db 100644 --- a/packages/ui-server/tests/ws-observability.test.ts +++ b/packages/ui-server/tests/ws-observability.test.ts @@ -604,7 +604,7 @@ describe("socket-level failures are observable", () => { }); describe("session persistence failures are observable", () => { - test("a failed catalog write warns with the session id instead of vanishing", () => { + test("a failed catalog write warns with the session id instead of vanishing", async () => { const observability = createRecordingObservability(); const db = createUiDb(":memory:"); db.close(); @@ -612,6 +612,8 @@ describe("session persistence failures are observable", () => { // Still non-throwing — the turn must survive — but no longer silent. expect(() => catalog.persistSessionStub("s9", "prompt", null, "fake")).not.toThrow(); + // The write settles off the event loop (#2850); its failure is reported then. + await catalog.settled!(); const [record] = observability.logs.find({ scope: "db", severity: "WARN" }); expect(record.body).toBe("session persistence failed"); From c876e0bb773855d79a3f17d92077dcafa16e272e Mon Sep 17 00:00:00 2001 From: Alain Schlesser Date: Sun, 25 Oct 2026 18:57:56 +0100 Subject: [PATCH 3/4] test(ui-server): read a turn's roadblocks once its slot has ended Roadblocks are classified after the journal completes and commit on the write worker; the session's slot ends only once they have (#2850). With the catalog's writes now on the same queue, reading them at the journal's complete frame raced their commit. Refs #2850 Co-Authored-By: Claude Opus 5.5 --- packages/ui-server/tests/ws-turn-journal.test.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/packages/ui-server/tests/ws-turn-journal.test.ts b/packages/ui-server/tests/ws-turn-journal.test.ts index 62eeaaf171..44578274cd 100644 --- a/packages/ui-server/tests/ws-turn-journal.test.ts +++ b/packages/ui-server/tests/ws-turn-journal.test.ts @@ -66,6 +66,9 @@ test("host admission, tool completion and terminal observation produce scoped re if (receipt.type !== "turn_journal") throw new Error("Missing receipt"); release(); await wait(() => frames().some(frame => frame?.type === "turn_journal" && frame.journal.state === "complete")); + // Roadblocks are classified after the journal completes; the turn's slot + // ends only once they have committed (#2850). + await wait(() => !app.wsHost.coordinator.isTurnActive()); const details = app.wsHost.journal!.details("sess-ithaca", receipt.turnId)!; expect(details.summary.policyHash, "JOURNAL_FROZEN_ADMISSION_POLICY").toMatch(/^[a-f0-9]{64}$/); expect(details.changes.filter(change => change.path === "notes/journal.md").map(change => app.wsHost.journal!.version("sess-ithaca", receipt.turnId, change.sequence, "after")!.bytes(1024).toString()), "JOURNAL_TOOL_AND_TERMINAL_VERSIONS").toEqual(["Athena\r\n", "Penelope\r\n"]); From ad017ddd85d7689e1184775bc0afdeb45c482523 Mon Sep 17 00:00:00 2001 From: Alain Schlesser Date: Sun, 25 Oct 2026 20:57:08 +0100 Subject: [PATCH 4/4] docs: re-point citations after merging main Refs #2850 Co-Authored-By: Claude Opus 5.5 --- docs/decisions/governed-writable-workers.md | 10 +++++----- docs/decisions/http-api-boundary.md | 8 ++++---- docs/decisions/session-principals.md | 6 +++--- docs/plans/async-collaboration-requirements.md | 6 +++--- docs/plans/async-collaboration.md | 6 +++--- docs/plans/navigation-recovery.md | 6 +++--- 6 files changed, 21 insertions(+), 21 deletions(-) diff --git a/docs/decisions/governed-writable-workers.md b/docs/decisions/governed-writable-workers.md index 8ca3167073..3f0724560a 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:12-89`; +(`createSessionProfiles`, `packages/ui-server/src/ws/session-profiles.ts:16-132`; `begin`, `packages/ui-server/src/brain/governed-turn.ts:141-306`). **Where a turn falls back to `strict`.** The launcher refuses to start @@ -163,7 +163,7 @@ still holds. **Voice.** Ruling 1 makes voice `governed` too. Voice keeps its narrower membership (`VOICE_ALLOWED_TOOLS`, `packages/ui-backend-claude/src/tool-policy.ts:120`): no Bash, no raw `Write`/`Edit`, no archive, and approval returns false -(`const apply = createBrainApplication`, `packages/ui-server/src/ws/bridge.ts:658-663`). +(`const apply = createBrainApplication`, `packages/ui-server/src/ws/bridge.ts:662-667`). A voice turn therefore writes only through curated capture and update, now with a journal behind it. The writable mount gives voice no tool it lacked. Pi still refuses voice turns. @@ -1072,7 +1072,7 @@ principal authority remain the caller's responsibility. The policy publisher creates, writes, fsyncs and removes its operation-owned staging name entirely inside that synchronous callback, between the two serialized observations (`await options.observeUserPolicy`, -`packages/ui-server/src/brain/policy-proposal.ts:198-225`). First publication +`packages/ui-server/src/brain/policy-proposal.ts:210-237`). First publication also removes the temporary hardlink before the final observation. No filename prefix is ignored; a similarly named real agent change is still retained. Failed staging preserves the original policy, while failed recording after @@ -1660,7 +1660,7 @@ The brain's git history stays the durable record of its content. - **The Activity recorder and its rollups** price those events under the [cost-tracking rules](cost-tracking.md). An unknown price is never counted as zero. The session total that counts unknown as zero - (`msg.costUsd ?? 0`, `packages/ui-server/src/ws/session-catalog.ts:368`) + (`msg.costUsd ?? 0`, `packages/ui-server/src/ws/session-catalog.ts:375`) must not feed any budget. **Enforcement:** @@ -2587,7 +2587,7 @@ every turn without a proven recipient: - **The handoff summary.** Drafting a handoff summary refuses when the policy has provider rules and the source chat's lineage does not cover every provider folder, before its history is read (`unprovenHistory`, - `packages/ui-server/src/ws/handoff.ts:303-306`). The run has no proven + `packages/ui-server/src/ws/handoff.ts:309-312`). The run has no proven recipient, so every provider folder is denied to it. A policy with a type selector always refuses: the type's folders follow the current taxonomy, which this check does not translate, and a remembered translation may be diff --git a/docs/decisions/http-api-boundary.md b/docs/decisions/http-api-boundary.md index d86f8d795b..f8c1be7bc3 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:860`), the API guard follows -public ceremonies (`authGuard`, `packages/ui-server/src/app.ts:881`), and the -WebSocket admission is separate (`"/ws"`, `packages/ui-server/src/app.ts:1070`). +(`createHealthRoutes`, `packages/ui-server/src/app.ts:864`), the API guard follows +public ceremonies (`authGuard`, `packages/ui-server/src/app.ts:885`), and the +WebSocket admission is separate (`"/ws"`, `packages/ui-server/src/app.ts:1075`). Static serving is conditional (`if (options.staticRoot)`, -`packages/ui-server/src/app.ts:1090-1111`). That order establishes access; +`packages/ui-server/src/app.ts:1095-1116`). 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 9ce28558bd..595ba87e82 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:866` | mounted **before** the guard at `authGuard(authMode`, `app.ts:881`; logout guards itself (`resolveCookiePrincipal(c, auth, deps.db)`, `auth.ts:808`) | -| Passkey management | `passkeyManagementRoutes`, `app.ts:891` | mounted **after** the guard ("Mount AFTER the auth guard") | -| WS guard | `"/ws"`, `app.ts:1070-1087` | origin → `isWsAuthorized` (returns a boolean, `auth.ts:221-237`) → capacity → upgrade | +| Auth routes | `authRoutes`, `app.ts:870` | mounted **before** the guard at `authGuard(authMode`, `app.ts:885`; logout guards itself (`resolveCookiePrincipal(c, auth, deps.db)`, `auth.ts:808`) | +| Passkey management | `passkeyManagementRoutes`, `app.ts:895` | mounted **after** the guard ("Mount AFTER the auth guard") | +| WS guard | `"/ws"`, `app.ts:1075-1092` | 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/plans/async-collaboration-requirements.md b/docs/plans/async-collaboration-requirements.md index eeedd9db76..e7753fb467 100644 --- a/docs/plans/async-collaboration-requirements.md +++ b/docs/plans/async-collaboration-requirements.md @@ -43,7 +43,7 @@ session. Three consequences: 1. **Background work cannot ask.** Cron runs (`sync`, `validate`, `maintain`, module jobs) have no human attached. `bridge.requestPermission` parks a promise that nobody will - resolve (`requestPermission: (req) => {`, `packages/ui-server/src/ws/bridge.ts:354-448`), so autonomous work is confined to + resolve (`requestPermission: (req) => {`, `packages/ui-server/src/ws/bridge.ts:358-452`), so autonomous work is confined to whatever is pre-approved, and anything requiring judgment is not attempted. 2. **Inbound material has no path.** A forwarded email, a shared link, a captured note has nowhere to land that the agent will act on later. The PWA share target stages into @@ -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:881`). + (`app.use("/api/*", authGuard(`, `packages/ui-server/src/app.ts:885`). 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 @@ -263,7 +263,7 @@ Stated before the requirements because five of them derive from it. threshold (~20s of the 30s lock budget). Below it nothing yields; above it the holder checkpoints/unwinds/releases. Test both edges and independent paths. - R26. **Abort-and-redo needs a checkpoint primitive.** `requestPermission` parks a bare - promise (`return new Promise((resolve) => {`, `packages/ui-server/src/ws/bridge.ts:398-456`) — while blocked on it the model cannot write anything, so + promise (`return new Promise((resolve) => {`, `packages/ui-server/src/ws/bridge.ts:402-460`) — while blocked on it the model cannot write anything, so "writes its findings, then aborts" has nowhere to run. The autonomous bridge must, in one server-side step: capture the checkpoint, create the Action and block the item, unwind without a live approval promise, preserving the tested timeout path's abort-then-drain order (`abortController.abort();`, `packages/ui-server/src/ws/run-session.ts:393-399`). Aborting does **not** undo completed tool side diff --git a/docs/plans/async-collaboration.md b/docs/plans/async-collaboration.md index c4fbf7d1ce..ce97e856af 100644 --- a/docs/plans/async-collaboration.md +++ b/docs/plans/async-collaboration.md @@ -28,7 +28,7 @@ than last. ## Problem Frame Collaboration is synchronous-only: background work cannot ask (`requestPermission` parks a -promise nobody resolves — (`requestPermission: (req) => {`, `packages/ui-server/src/ws/bridge.ts:354-448`)), inbound material +promise nobody resolves — (`requestPermission: (req) => {`, `packages/ui-server/src/ws/bridge.ts:358-452`)), inbound material has no path that survives until the user is present, and no decision accumulates into standing authority. See origin for the full frame. @@ -131,10 +131,10 @@ 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:881`) follows public routes; +- **Auth mounting:** (`app.use("/api/*", authGuard(`, `packages/ui-server/src/app.ts:885`) follows public routes; (`export function authGuard(`, `packages/ui-server/src/middleware/auth.ts:189-249`) 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:354-448`). Timeout unwind is +- **Permission parking:** (`requestPermission: (req) => {`, `packages/ui-server/src/ws/bridge.ts:358-452`). Timeout unwind is (`abortController.abort();`, `packages/ui-server/src/ws/run-session.ts:393-399`). Durable escalation must checkpoint before unwinding; the existing ordinary bridge does not do that. - **Tool enforcement:** (`const enforcementHook: HookCallback`, `packages/ui-backend-claude/src/permission-hooks.ts:200-229`) closes measured bypasses. diff --git a/docs/plans/navigation-recovery.md b/docs/plans/navigation-recovery.md index 5a16010028..4d0fdbb608 100644 --- a/docs/plans/navigation-recovery.md +++ b/docs/plans/navigation-recovery.md @@ -385,9 +385,9 @@ 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:621-675`). +- Host resume: (`case "session_resume": {`, `packages/ui-server/src/ws/dispatch.ts:628-682`). -- Catalog activity: (`persistSessionStub(sessionId, promptText, providerId, backendId) {`, `packages/ui-server/src/ws/session-catalog.ts:339-378`). +- Catalog activity: (`persistSessionStub(sessionId, promptText, providerId, backendId) {`, `packages/ui-server/src/ws/session-catalog.ts:369-376`). - Session summary shape: (`export interface ChatSession {`, `packages/ui-sdk/src/protocol.ts:1568-1584`). @@ -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:881`). +- API authorization: (`app.use("/api/*", authGuard(authMode, auth, db));`, `packages/ui-server/src/app.ts:885`). - Queued acceptance: (`// Queue it as the session's next turn; report queued immediately.`, `packages/ui-server/src/ws/run-session.ts:1098-1143`).