From 3d618f5c90ef38d82ebc8907a60108f29a1569b8 Mon Sep 17 00:00:00 2001 From: Alain Schlesser Date: Mon, 26 Oct 2026 16:02:10 +0100 Subject: [PATCH 1/4] fix(ui-server): move account administration writes offloop --- .changeset/account-admin-write-worker.md | 5 + docs/decisions/action-notifications.md | 2 +- docs/decisions/claude-code-runtime.md | 4 +- docs/decisions/hardening.md | 2 +- docs/decisions/http-api-boundary.md | 8 +- docs/decisions/reference-source-grants.md | 4 +- docs/decisions/scheduled-tasks.md | 2 +- docs/decisions/session-principals.md | 18 +- docs/decisions/sqlite-write-worker.md | 25 +- docs/inbox-notifications.md | 2 +- .../plans/async-collaboration-requirements.md | 4 +- docs/plans/async-collaboration.md | 6 +- docs/plans/navigation-recovery.md | 2 +- docs/plans/storage-accounting.md | 2 +- .../tests/browser/offline/fault-network.ts | 2 +- .../ui-server/src/activity/push-sender.ts | 88 ++---- .../ui-server/src/activity/push-writes.ts | 46 +++ packages/ui-server/src/activity/runtime.ts | 6 +- packages/ui-server/src/app.ts | 4 +- packages/ui-server/src/db/account-writes.ts | 97 +++++++ packages/ui-server/src/db/principal-writes.ts | 4 +- .../src/middleware/account-partition.ts | 4 +- packages/ui-server/src/middleware/auth.ts | 13 +- packages/ui-server/src/middleware/passkeys.ts | 54 +--- .../ui-server/src/middleware/principals.ts | 17 +- packages/ui-server/src/routes/push.ts | 12 +- packages/ui-server/src/storage/sampling.ts | 2 +- packages/ui-server/src/ws/host.ts | 8 +- .../tests/account-writes-offloop.test.ts | 273 ++++++++++++++++++ 29 files changed, 544 insertions(+), 172 deletions(-) create mode 100644 .changeset/account-admin-write-worker.md create mode 100644 packages/ui-server/src/activity/push-writes.ts create mode 100644 packages/ui-server/src/db/account-writes.ts create mode 100644 packages/ui-server/tests/account-writes-offloop.test.ts diff --git a/.changeset/account-admin-write-worker.md b/.changeset/account-admin-write-worker.md new file mode 100644 index 0000000000..5ad9b876fe --- /dev/null +++ b/.changeset/account-admin-write-worker.md @@ -0,0 +1,5 @@ +--- +"@schlessera/brain-ui-server": patch +--- + +Commit account and security administration, passkeys, reference grants and push registration on the SQLite write worker at FULL durability, so their fsyncs do not block requests. Revocation responses await the durable revocation, and concurrent VAPID first uses share one stored key pair. diff --git a/docs/decisions/action-notifications.md b/docs/decisions/action-notifications.md index 2a2be2398e..15ed89cb25 100644 --- a/docs/decisions/action-notifications.md +++ b/docs/decisions/action-notifications.md @@ -181,7 +181,7 @@ incrementing an Action count (`function createIntent(input: {`, `packages/ui-server/src/activity/notify.ts:94-130`). The sender processes run-bound activity rows and settles one intent from a device pass; that is not the required per-episode/per-destination Action history -(`async deliverPending(notifier) {`, `packages/ui-server/src/activity/push-sender.ts:208-282`). +(`async deliverPending(notifier) {`, `packages/ui-server/src/activity/push-sender.ts:172-246`). Reuse concrete subscriptions, authorization, revocation and bounded backoff, but add the Actions-aware aggregate and durable constituent/destination records. This decision authorizes no new channel, extension seam or daemon. diff --git a/docs/decisions/claude-code-runtime.md b/docs/decisions/claude-code-runtime.md index b59ebeee95..8279c9757b 100644 --- a/docs/decisions/claude-code-runtime.md +++ b/docs/decisions/claude-code-runtime.md @@ -64,7 +64,7 @@ next to it. of this binary. The only version probe in the server is for the `brain` CLI (`Probe the selected content CLI`, `packages/ui-server/src/brain/client.ts:142-210`, called at - `probeBrainCliVersion(config.brainPath`, `packages/ui-server/src/app.ts:436`). + `probeBrainCliVersion(config.brainPath`, `packages/ui-server/src/app.ts:437`). `brain doctor` runs `claude mcp list` from `PATH` (`which("claude")`, `packages/core/src/cli/commands/doctor.ts:503-507`) — the user's own Claude Code on their own machine, to check the MCP registration, not the server's @@ -212,7 +212,7 @@ unnecessary. `node_modules/@anthropic-ai/claude-agent-sdk/sdk.mjs`. The boot probe below turns either into a refusal to start, for the same reason a missing backend refuses to boot (`A missing (or unrecognized) agent backend`, - `packages/ui-server/src/app.ts:374-379`). + `packages/ui-server/src/app.ts:375-380`). - **The published range is still a caret, and that bounds what this repo can guarantee.** `@schlessera/brain-backend-claude` depends on `^0.3.241`, and a host resolves it in its own lockfile. A host can bump the SDK — and so the diff --git a/docs/decisions/hardening.md b/docs/decisions/hardening.md index 3bb00760ce..4dae5852b8 100644 --- a/docs/decisions/hardening.md +++ b/docs/decisions/hardening.md @@ -46,7 +46,7 @@ instead, and treat 8 as the question it was answering. epoch. Logout and passkey revocation bump it **and close every open WebSocket** (new `ClientSet.closeAll`). **The bump requires a valid epoch-bearing session**: logout is a public route today - (`/auth/logout`, `packages/ui-server/src/middleware/auth.ts:797`, + (`/auth/logout`, `packages/ui-server/src/middleware/auth.ts:796`, mounted before the guard), and a public epoch bump would let anyone log the owner out of everything in a loop. The epoch accessor is strict, not the fail-open settings getter: first initialization is distinguished from diff --git a/docs/decisions/http-api-boundary.md b/docs/decisions/http-api-boundary.md index df021d4599..40e88e82e1 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:872`), the API guard follows -public ceremonies (`authGuard`, `packages/ui-server/src/app.ts:893`), and the -WebSocket admission is separate (`"/ws"`, `packages/ui-server/src/app.ts:1083`). +(`createHealthRoutes`, `packages/ui-server/src/app.ts:874`), the API guard follows +public ceremonies (`authGuard`, `packages/ui-server/src/app.ts:895`), and the +WebSocket admission is separate (`"/ws"`, `packages/ui-server/src/app.ts:1085`). Static serving is conditional (`if (options.staticRoot)`, -`packages/ui-server/src/app.ts:1103-1124`). That order establishes access; +`packages/ui-server/src/app.ts:1105-1126`). 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/reference-source-grants.md b/docs/decisions/reference-source-grants.md index 054be5c0e0..4803485546 100644 --- a/docs/decisions/reference-source-grants.md +++ b/docs/decisions/reference-source-grants.md @@ -39,7 +39,7 @@ Only a live owner can create a grant, and only for a live owner or agent principal. Cron's system principal and ambient principals cannot hold one. The routes sit inside the principal-management family, so they inherit its password-mode and owner checks -(`app.get("/auth/principals/:id/reference-grants"`, `packages/ui-server/src/middleware/principals.ts:219-228`). +(`app.get("/auth/principals/:id/reference-grants"`, `packages/ui-server/src/middleware/principals.ts:216-225`). Owner sessions are principals too: every login creates one, and a grant to an owner session ends when that session does. @@ -78,7 +78,7 @@ anonymous success there says nothing about who may read the host path. Revocation is checked, not cached. Every new mount and every step that could publish private source asks for the live covering grant again. The revoke routes also report the lost authority synchronously -(`app.delete("/auth/principals/:id/reference-grants/:grantId"`, `packages/ui-server/src/middleware/principals.ts:275-296`), +(`app.delete("/auth/principals/:id/reference-grants/:grantId"`, `packages/ui-server/src/middleware/principals.ts:272-293`), so the acquisition layer can cancel in-flight private work rather than wait for its next check. Revocation cannot retract bytes a reader has already read. diff --git a/docs/decisions/scheduled-tasks.md b/docs/decisions/scheduled-tasks.md index 551a37ef10..10e0ac9f4c 100644 --- a/docs/decisions/scheduled-tasks.md +++ b/docs/decisions/scheduled-tasks.md @@ -23,7 +23,7 @@ Source inspected at `b5c30c180df5584584f261b446c4d440380d2441`: (`else if (item.status === "scheduled"`, `packages/ui-server/src/inbox/actions.ts:196-197`). The runtime ticks every minute (`INBOX_TICK_MS`, `packages/ui-server/src/inbox/runtime.ts:11-13`). - Production app wiring supplies recovery and budgets, without a dispatcher - (`// Recovery/heartbeat only.`, `packages/ui-server/src/app.ts:497-513`). + (`// Recovery/heartbeat only.`, `packages/ui-server/src/app.ts:499-515`). A scripted nonempty ready-item test still reports no claim, attempt or Activity root. That proves disabled wiring, not safe execution of a schedule. - Headless plumbing checks backend support, a usable principal, lifetime and diff --git a/docs/decisions/session-principals.md b/docs/decisions/session-principals.md index dff313c435..2cf6a63b05 100644 --- a/docs/decisions/session-principals.md +++ b/docs/decisions/session-principals.md @@ -66,7 +66,7 @@ incident; neither should be the standing answer. - **Scopes / least privilege per route.** The one exception is R9's kind check on the principal-management routes — a single check on a single route family, the same shape as the existing password-disabled guard (`deps.passwordDisabled`, - `packages/ui-server/src/middleware/auth.ts:698`), not a scope system. + `packages/ui-server/src/middleware/auth.ts:697`), not a scope system. - **Multi-user.** One owner, several principals. - **Bearer tokens.** Additive later and cheap (see Q5); not in this plan. - **Containment of a compromised principal.** See the Summary. @@ -96,16 +96,16 @@ 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:878` | mounted **before** the guard at `authGuard(authMode`, `app.ts:893`; logout guards itself (`resolveCookiePrincipal(c, auth, deps.db)`, `auth.ts:799`) | -| Passkey management | `passkeyManagementRoutes`, `app.ts:903` | mounted **after** the guard ("Mount AFTER the auth guard") | -| WS guard | `"/ws"`, `app.ts:1083-1100` | origin → `isWsAuthorized` (returns a boolean, `auth.ts:221-237`) → capacity → upgrade | +| Auth routes | `authRoutes`, `app.ts:880` | mounted **before** the guard at `authGuard(authMode`, `app.ts:895`; logout guards itself (`resolveCookiePrincipal(c, auth, deps.db)`, `auth.ts:798`) | +| Passkey management | `passkeyManagementRoutes`, `app.ts:905` | mounted **after** the guard ("Mount AFTER the auth guard") | +| WS guard | `"/ws"`, `app.ts:1085-1102` | 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 | | Follow-up queue | `QueuedFollowUp`, `ws/turns.ts:61-103`; `slot.queue.push(entry)`, `run-session.ts:1121` | queued entries re-mint `turnId` in the same slot and may come from another socket | -| Password login | `acquirePasswordVerification(key)`, `auth.ts:740` | argon2id verify, failure counting, in-flight reservation, then `issueSessionCookie` | -| Passkey login | `issueLoginSession`, `packages/ui-server/src/middleware/passkeys.ts:448` | assertion verified, then `issueSessionCookie`; `row.id` is in scope | -| Passkey **registration** | `INSERT INTO passkey_credentials`, `passkeys.ts:547` | inserts a credential; calls **neither** helper | +| Password login | `acquirePasswordVerification(key)`, `auth.ts:739` | argon2id verify, failure counting, in-flight reservation, then `issueSessionCookie` | +| Passkey login | `issueLoginSession`, `packages/ui-server/src/middleware/passkeys.ts:441` | assertion verified, then `issueSessionCookie`; `row.id` is in scope | +| Passkey **registration** | `INSERT INTO passkey_credentials`, `packages/ui-server/src/db/account-writes.ts:31-36` | inserts a credential; calls **neither** helper | | Passkey delete | `passkeys.ts:578-586` | deletes the credential, then a **global** bump | | Sockets | `ClientSet`, `ws/clients.ts:41` | `ClientSet` keyed on `ws.raw` — hono mints a fresh `WSContext` per callback | | Activity subscriptions | `activity/stream.ts:153` | a **separate** registry keyed on the wrapper, not `ws.raw` | @@ -163,7 +163,7 @@ edited without breaking the signature. must not be able to drive database reads with arbitrary ids. The corrupt-row throw sits after signature verification and outside the cookie-parser catch, exactly as today's code does (`Deliberately after signature`, -`auth.ts:416-418`) — so a corrupt row 500s only that principal's requests +`auth.ts:415-417`) — so a corrupt row 500s only that principal's requests instead of everyone's. **4. Principals carry their lineage, and passkey deletion uses it.** @@ -216,7 +216,7 @@ nothing is backfilled. > example does not describe the current root span. **8. Ambient modes keep the identity they already have.** `proxy` mode reads an -upstream user header (`auth.proxyAuthHeader`, `auth.ts:476`) — that becomes the +upstream user header (`auth.proxyAuthHeader`, `auth.ts:475`) — that becomes the ambient principal's label, bounded like `sanitizeLabel`, `passkeys.ts:273-275`. `tailscale` carries the client IP; `none` is fixed. Minting and revocation are disabled in these modes. Cron is attributed to an explicit system principal, never to the owner. diff --git a/docs/decisions/sqlite-write-worker.md b/docs/decisions/sqlite-write-worker.md index a770f75642..d95b564cdc 100644 --- a/docs/decisions/sqlite-write-worker.md +++ b/docs/decisions/sqlite-write-worker.md @@ -183,8 +183,25 @@ are in [the governed-workers decision](governed-writable-workers.md#admitted-obj (`packages/ui-server/src/db/principal-writes.ts`). A touch is never waited for and one pending is not submitted again; an ambient principal that is current is read at once; a login answers once its row has committed, and - its refusals keep their classes. Revocations and administrative principal - writes still commit on the app connection. + its refusals keep their classes. Account administration uses the same worker, + as described below. +- **Account and security administration followed** (#3002, within #2850). + Delegated-principal creation and revocation, passkey registration, counters, + labels and deletion, reference grants and the account partition seed commit + on the worker (`packages/ui-server/src/db/account-writes.ts`). The existing + transaction bodies and modes stay together; passkey deletion still commits + before its credential's principal revocation. Every route awaits its writes + before answering or invalidating runtime authority, so every authentication + after a revocation response sees the committed revocation. Delegated minting + and reference grants recheck authority inside their original transactions; + refusals keep their classes and HTTP statuses. Boot awaits the partition seed + before constructing its synchronous key reader. Push subscription, removal + and unbinding, and first-use VAPID creation run on the same queue + (`packages/ui-server/src/activity/push-writes.ts`). Each VAPID caller checks + for the stored pair on that connection, so racing delivery passes and the + key route share one insert and its read-back. Delivery still joins devices + to live principals while runtime unbinding is pending. Standalone callers + retain the synchronous procedures on their own connection. - **The Activity notifier followed** (#2850). Its tick (intent detection under its cursor and the stuck-run watchdog), delivery status, acknowledgement and retention, and push delivery's subscription @@ -262,9 +279,7 @@ are in [the governed-workers decision](governed-writable-workers.md#admitted-obj creates the root identity on the app connection during boot, before serving; the unwired admission service and cron execution retain their own writes. - **Other app-connection writers stay on the loop for now**: retained - retries, drafts, policy proposals, storage notices, push - subscriptions and VAPID keys, principal revocation and administration, - passkeys and settings. Each commits at + retries, drafts, policy proposals, storage notices 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/inbox-notifications.md b/docs/inbox-notifications.md index 18f751448b..840de4686f 100644 --- a/docs/inbox-notifications.md +++ b/docs/inbox-notifications.md @@ -42,7 +42,7 @@ deadline opens the next window. Thread identity is kept on each constituent, not in the grouping key. Unmarked below-cutoff decisions and FYIs never join a window. Delivery runs after activity intents on the same reentrancy guard -(`deliverActions`, `packages/ui-server/src/activity/push-sender.ts:327-378`). +(`deliverActions`, `packages/ui-server/src/activity/push-sender.ts:291-342`). Each destination is evaluated on its own. `beginAttempt()` rechecks, in one transaction, that the subscription is still bound to the principal and the principal is usable, that the device reported a usable zone and is outside diff --git a/docs/plans/async-collaboration-requirements.md b/docs/plans/async-collaboration-requirements.md index aeebe2b7ee..ab93991bd1 100644 --- a/docs/plans/async-collaboration-requirements.md +++ b/docs/plans/async-collaboration-requirements.md @@ -226,7 +226,7 @@ Stated before the requirements because five of them derive from it. that is the shape to copy, including its `close()` lifecycle. - R19. **The cron backstop has independent authorization before the general guard.** Mount it on the existing listener before - (`app.use("/api/*", authGuard(`, `packages/ui-server/src/app.ts:893`). + (`app.use("/api/*", authGuard(`, `packages/ui-server/src/app.ts:895`). 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 @@ -387,7 +387,7 @@ Stated before the requirements because five of them derive from it. to `failure|completion|stuck`, requires an activity `run_id`, coalesces only by dropping a later same-tag intent without updating a count (`function createIntent(input: {`, `packages/ui-server/src/activity/notify.ts:94-130`), and the sender emits one push per - pending row (`async deliverPending(notifier) {`, `packages/ui-server/src/activity/push-sender.ts:208-255`). Reuse subscriptions, retry budget, and delivery + pending row (`async deliverPending(notifier) {`, `packages/ui-server/src/activity/push-sender.ts:172-219`). Reuse subscriptions, retry budget, and delivery status; add an Actions-aware aggregate carrying group key, count, priority, quiet-hours eligibility, and an Actions deep link. The [2026-10-02 Action notification decision](../decisions/action-notifications.md) diff --git a/docs/plans/async-collaboration.md b/docs/plans/async-collaboration.md index 472c7c2344..7cb72b60c6 100644 --- a/docs/plans/async-collaboration.md +++ b/docs/plans/async-collaboration.md @@ -131,8 +131,8 @@ 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:893`) follows public routes; - (`export function authGuard(`, `packages/ui-server/src/middleware/auth.ts:185-245`) binds principals in each auth mode. +- **Auth mounting:** (`app.use("/api/*", authGuard(`, `packages/ui-server/src/app.ts:895`) follows public routes; + (`export function authGuard(`, `packages/ui-server/src/middleware/auth.ts:184-244`) 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:361-455`). Timeout unwind is (`abortController.abort();`, `packages/ui-server/src/ws/run-session.ts:393-399`). Durable escalation must checkpoint @@ -151,7 +151,7 @@ system gate. It is not a prerequisite to rebuild or a new independent package. Reservations must cover in-flight work, not only this retrospective sum. - **Notifications:** (`CREATE TABLE IF NOT EXISTS notification_intents (`, `packages/ui-server/migrations/007_activity.sql:99-114`), (`function createIntent(input: {`, `packages/ui-server/src/activity/notify.ts:94-130`) and - (`async deliverPending(notifier) {`, `packages/ui-server/src/activity/push-sender.ts:208-255`) are run-bound and do not + (`async deliverPending(notifier) {`, `packages/ui-server/src/activity/push-sender.ts:172-219`) are run-bound and do not maintain an Actions aggregate count. - **Share provenance/limits:** (`const result = await stageShareAt(`, `packages/ui-server/src/inbox/intake.ts:101`) assigns the source in server code. (`export const SHARE_MAX_TEXT_BYTES =`, `packages/ui-sdk/src/protocol.ts:2348`) bounds text, not binary uploads; diff --git a/docs/plans/navigation-recovery.md b/docs/plans/navigation-recovery.md index 04a2a75c4d..0a3f2f94b8 100644 --- a/docs/plans/navigation-recovery.md +++ b/docs/plans/navigation-recovery.md @@ -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:144-282`). -- API authorization: (`app.use("/api/*", authGuard(authMode, auth, db));`, `packages/ui-server/src/app.ts:893`). +- API authorization: (`app.use("/api/*", authGuard(authMode, auth, db));`, `packages/ui-server/src/app.ts:895`). - 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/docs/plans/storage-accounting.md b/docs/plans/storage-accounting.md index 48efe87c5b..650d9bb626 100644 --- a/docs/plans/storage-accounting.md +++ b/docs/plans/storage-accounting.md @@ -67,7 +67,7 @@ Source anchors, checked against the reviewed commit: The completed #94 and #398 provide existing stats/maintenance foundations. - C: snapshot fields (`snapshotOf`, `packages/core/src/lib/stats-history.ts:132-172`), thinning (`thin`, `packages/core/src/lib/stats-history.ts:232-244`). -- D: effective path/journal resolution (`createApp`, `packages/ui-server/src/app.ts:295-331`), +- D: effective path/journal resolution (`createApp`, `packages/ui-server/src/app.ts:296-332`), WAL store (`createUiDb`, `packages/ui-server/src/db/client.ts:27-44`), and private locations (`statePaths`, `packages/ui-server/src/brain/operational-state.ts:108-114`). - E: live/archive paths (`transcriptPaths`, `packages/ui-backend-claude/src/transcript-archive.ts:30-39`). diff --git a/packages/ui-react/tests/browser/offline/fault-network.ts b/packages/ui-react/tests/browser/offline/fault-network.ts index 10e61dd4fe..48911c2b60 100644 --- a/packages/ui-react/tests/browser/offline/fault-network.ts +++ b/packages/ui-react/tests/browser/offline/fault-network.ts @@ -11,7 +11,7 @@ * - `expireAuth()`: the probe and every request answer 401, open sockets close * with 1008 (the host's expiry and revocation close: * `SESSION_EXPIRED_CLOSE_CODE`, `packages/ui-server/src/ws/host.ts:54`; - * `SESSION_REVOKED_CLOSE_CODE`, `packages/ui-server/src/middleware/auth.ts:66`), + * `SESSION_REVOKED_CLOSE_CODE`, `packages/ui-server/src/middleware/auth.ts:65`), * and new sockets fail their handshake, as an upgrade without a session does. * * Pass `network.request` to `createBrainUiRoot({ request })`. The global diff --git a/packages/ui-server/src/activity/push-sender.ts b/packages/ui-server/src/activity/push-sender.ts index c9c1da78a5..ce9f2833da 100644 --- a/packages/ui-server/src/activity/push-sender.ts +++ b/packages/ui-server/src/activity/push-sender.ts @@ -15,6 +15,8 @@ * re-enable prompt when its subscription check disagrees with the server. */ import type { Database } from "bun:sqlite"; +import type { SqliteWriter } from "../db/write-worker.js"; +import { createPushWrites, setup as inlinePushWrites } from "./push-writes.js"; import type { PushWrites } from "./notify-writes.js"; import type { Logger } from "@opentelemetry/api-logs"; import webpush from "web-push"; @@ -59,7 +61,16 @@ export interface PushSender { deliverStorage(notices: StorageNotifier, now?: number, signal?: AbortSignal): Promise; } + +/** Hosted lifecycle writes settle after their FULL worker commit; standalone reads stay synchronous. */ +type LifecycleWrite = "publicKey" | "subscribe" | "unsubscribe" | "unbindPrincipal"; +export type AwaitablePushSender = Omit & { + [K in LifecycleWrite]: (...args: Parameters) => ReturnType | Promise>; +}; + export interface CreatePushSenderOptions { + /** Shared hosted write connection; absent for standalone synchronous callers. */ + writer?: SqliteWriter; /** Injection seam for tests — the real webpush.sendNotification otherwise. */ send?: ( subscription: { endpoint: string; keys: { p256dh: string; auth: string } }, @@ -80,10 +91,10 @@ export interface CreatePushSenderOptions { * delivering guard (the tick's reentrancy flag) forever. */ const SEND_TIMEOUT_MS = 10_000; -export function createPushSender( - db: Database, - options: CreatePushSenderOptions = {} -): PushSender { +export function createPushSender(db: Database, options: CreatePushSenderOptions & { writer: SqliteWriter }): AwaitablePushSender; +export function createPushSender(db: Database, options?: CreatePushSenderOptions & { writer?: undefined }): PushSender; +export function createPushSender(db: Database, options: CreatePushSenderOptions): AwaitablePushSender; +export function createPushSender(db: Database, options: CreatePushSenderOptions = {}): AwaitablePushSender { const log = options.log; const writes: PushWrites = options.writes ?? { async touch(endpoint, at, principalId) { @@ -102,19 +113,9 @@ export function createPushSender( { vapidDetails: opts.vapidDetails, timeout: opts.timeout } )); - // Generate-once VAPID keys. Concurrent first boots are safe: the INSERT is - // OR IGNORE and the read-back wins. - function vapid(): { publicKey: string; privateKey: string } { - const row = db - .query("SELECT public_key AS publicKey, private_key AS privateKey FROM vapid_keys WHERE id = 1") - .get() as { publicKey: string; privateKey: string } | null; - if (row) return row; - const generated = webpush.generateVAPIDKeys(); - db.query( - "INSERT OR IGNORE INTO vapid_keys (id, public_key, private_key, created_at) VALUES (1, ?, ?, ?)" - ).run(generated.publicKey, generated.privateKey, Date.now()); - return vapid(); - } + // The worker serializes each first-use check and insert, including races + // between delivery passes and the public-key route. + const admin = options.writer ? createPushWrites(db, options.writer) : inlinePushWrites(db); function mapRows(dbRows: any[]): PushSubscriptionRow[] { return ( @@ -160,50 +161,13 @@ export function createPushSender( return { publicKey() { - return vapid().publicKey; - }, - - subscribe(sub, principalId, label) { - const now = Date.now(); - return db.query( - `INSERT INTO push_subscriptions - (endpoint, p256dh, auth, label, created_at, principal_id) - SELECT ?, ?, ?, ?, ?, id FROM principals - WHERE id = ? AND revoked_at IS NULL - AND typeof(expires_at) = 'integer' AND expires_at > ? - ON CONFLICT(endpoint) DO UPDATE SET - p256dh = excluded.p256dh, - auth = excluded.auth, - label = excluded.label, - principal_id = excluded.principal_id` - ).run( - sub.endpoint, - sub.keys.p256dh, - sub.keys.auth, - label ?? null, - now, - principalId, - now - ).changes > 0; - }, - - unsubscribe(endpoint, principalId) { - return db - .query( - "DELETE FROM push_subscriptions WHERE endpoint = ? AND principal_id = ?" - ) - .run(endpoint, principalId).changes > 0; + const keys = admin.vapid(); + return keys instanceof Promise ? keys.then(row => row.publicKey) : keys.publicKey; }, - + subscribe: (sub, principalId, label) => admin.subscribe(sub, principalId, label, Date.now()), + unsubscribe: (endpoint, principalId) => admin.unsubscribe(endpoint, principalId), subscriptions: rows, - - unbindPrincipal(principalId) { - return db - .query( - "UPDATE push_subscriptions SET principal_id = NULL WHERE principal_id = ?" - ) - .run(principalId).changes; - }, + unbindPrincipal: principalId => admin.unbindPrincipal(principalId), async deliverPending(notifier) { const pending = notifier.pending(); @@ -214,7 +178,7 @@ export function createPushSender( // if they are fresh, and the boot sweep can retry. return 0; } - const keys = vapid(); + const keys = await admin.vapid(); let attempts = 0; for (const intent of pending) { @@ -299,7 +263,7 @@ export function createPushSender( // before its transport call. Authority/zone/time are rechecked per device. const subs = deliverableRows(at ?? Date.now()); if (!subs.length) return 0; - const keys = vapid(); + const keys = await admin.vapid(); for (const sub of subs) { if (signal?.aborted) break; const attempt = notices.begin(sub.endpoint, at ?? Date.now()); @@ -335,7 +299,7 @@ export function createPushSender( if (attempt) planned.push({ sub, attempt }); } if (planned.length === 0) return 0; - const keys = vapid(); + const keys = await admin.vapid(); const results = await Promise.allSettled( planned.map(({ sub, attempt }) => send( diff --git a/packages/ui-server/src/activity/push-writes.ts b/packages/ui-server/src/activity/push-writes.ts new file mode 100644 index 0000000000..d5f7cafc4b --- /dev/null +++ b/packages/ui-server/src/activity/push-writes.ts @@ -0,0 +1,46 @@ +/** Push registration, revocation and lazy VAPID creation off the app connection (#3002). */ +import type { Database } from "bun:sqlite"; +import webpush from "web-push"; +import { procedureCaller, type SqliteWriter } from "../db/write-worker.js"; + +export const url = import.meta.url; +type Subscription = { endpoint: string; keys: { p256dh: string; auth: string } }; + +export function setup(db: Database) { + const readKeys = () => db.query("SELECT public_key AS publicKey, private_key AS privateKey FROM vapid_keys WHERE id = 1") + .get() as { publicKey: string; privateKey: string } | null; + return { + vapid() { + const row = readKeys(); + if (row) return row; + const generated = webpush.generateVAPIDKeys(); + db.query("INSERT OR IGNORE INTO vapid_keys (id, public_key, private_key, created_at) VALUES (1, ?, ?, ?)") + .run(generated.publicKey, generated.privateKey, Date.now()); + return readKeys()!; + }, + subscribe(sub: Subscription, principalId: string, label: string | undefined, now: number) { + return db.query(`INSERT INTO push_subscriptions (endpoint, p256dh, auth, label, created_at, principal_id) + SELECT ?, ?, ?, ?, ?, id FROM principals WHERE id = ? AND revoked_at IS NULL + AND typeof(expires_at) = 'integer' AND expires_at > ? + ON CONFLICT(endpoint) DO UPDATE SET p256dh = excluded.p256dh, auth = excluded.auth, + label = excluded.label, principal_id = excluded.principal_id`) + .run(sub.endpoint, sub.keys.p256dh, sub.keys.auth, label ?? null, now, principalId, now).changes > 0; + }, + unsubscribe(endpoint: string, principalId: string) { + return db.query("DELETE FROM push_subscriptions WHERE endpoint = ? AND principal_id = ?").run(endpoint, principalId).changes > 0; + }, + unbindPrincipal(principalId: string) { + return db.query("UPDATE push_subscriptions SET principal_id = NULL WHERE principal_id = ?").run(principalId).changes; + }, + }; +} + +export function createPushWrites(db: Database, writer: SqliteWriter) { + const call = procedureCaller(db, { url, setup }, undefined, writer); + return { + vapid: () => call["vapid"]>>("vapid"), + subscribe: (sub: Subscription, principalId: string, label: string | undefined, now: number) => call("subscribe", sub, principalId, label, now), + unsubscribe: (endpoint: string, principalId: string) => call("unsubscribe", endpoint, principalId), + unbindPrincipal: (principalId: string) => call("unbindPrincipal", principalId), + }; +} diff --git a/packages/ui-server/src/activity/runtime.ts b/packages/ui-server/src/activity/runtime.ts index 00d88b46e6..c02f42149d 100644 --- a/packages/ui-server/src/activity/runtime.ts +++ b/packages/ui-server/src/activity/runtime.ts @@ -19,7 +19,7 @@ import type { SqliteWriter } from "../db/write-worker.js"; import { createActivityStream, type ActivityStream } from "./stream.js"; import { createOffloopNotifier } from "./notify-writes.js"; import { createActivityNotifier, type ActivityNotifier } from "./notify.js"; -import { createPushSender, type PushSender } from "./push-sender.js"; +import { createPushSender, type AwaitablePushSender } from "./push-sender.js"; import { createActionNotifier } from "../inbox/notify.js"; import { createOffloopActionNotifier, type AwaitableActionNotifier } from "../inbox/notify-writes.js"; import { digestRetentionFloor } from "./digest.js"; @@ -45,7 +45,7 @@ export interface ActivityRuntime { store: AwaitableActivityStore; stream: ActivityStream; notifier: ActivityNotifier; - pushSender: PushSender; + pushSender: AwaitablePushSender; /** Durable Action notices: windows, per-destination attempts, digest coverage. */ actionNotices: AwaitableActionNotifier; /** What the server knows about the agent runtime (boot probe, last turn, last auth failure). */ @@ -100,7 +100,7 @@ export function createActivityRuntime( isWatched: (scope) => stream.isWatched(scope), log, }); - const pushSender = createPushSender(db, { log, ...(offloop ? { writes: offloop.push } : {}) }); + const pushSender = createPushSender(db, { log, writer: deps.writer, ...(offloop ? { writes: offloop.push } : {}) }); const offloopActions = deps.writer ? createOffloopActionNotifier(db, deps.writer) : undefined; const actionNotices = offloopActions?.notices ?? createActionNotifier(db); diff --git a/packages/ui-server/src/app.ts b/packages/ui-server/src/app.ts index 63667515be..3915f45ea5 100644 --- a/packages/ui-server/src/app.ts +++ b/packages/ui-server/src/app.ts @@ -92,6 +92,7 @@ import { createModuleRoutes } from "./routes/modules.js"; import { createUiDb } from "./db/client.js"; import { openSqliteWriter, procedureCaller } from "./db/write-worker.js"; import * as governedTurnWrites from "./brain/governed-turn-writes.js"; +import { accountWrites, registerAccountWrites } from "./db/account-writes.js"; import { registerPrincipalWrites } from "./db/principal-writes.js"; import { JOURNAL_SCHEMA, journalDatabasePath } from "./db/journal-database.js"; import { createInboxBridgeTools } from "./inbox/bridge-tools.js"; @@ -457,6 +458,7 @@ export async function createApp(options: CreateAppOptions = {}): Promise writer.close()); reconciled(hostExecution.attach(db, writer)); @@ -893,7 +895,7 @@ export async function createApp(options: CreateAppOptions = {}): Promise { const accountKey = accountKeys(c.get("principal")); return c.json(accountKey ? { vpn: true, accountKey } : { vpn: true }); diff --git a/packages/ui-server/src/db/account-writes.ts b/packages/ui-server/src/db/account-writes.ts new file mode 100644 index 0000000000..79d48481e9 --- /dev/null +++ b/packages/ui-server/src/db/account-writes.ts @@ -0,0 +1,97 @@ +/** Account/security administration on the shared FULL write worker (#3002). */ +import type { Database } from "bun:sqlite"; +import { procedureCaller, type SqliteWriter, type WireError } from "./write-worker.js"; +import { createPrincipal, revokePrincipal, revokeAllPrincipals, revokeByCredential, PrincipalLimitError, + PrincipalCredentialNotFoundError, PrincipalCreatorNotUsableError, type CreatePrincipalInput } from "./principals.js"; +import { createReferenceGrant, revokeReferenceGrant, ReferenceRepositoryError, ReferenceGrantPrincipalError, + ReferenceGrantDuplicateError, ReferenceGrantLimitError } from "./reference-grants.js"; +import { accountPartitionSeed } from "../middleware/account-partition.js"; + +export const url = import.meta.url; + +export interface PasskeyRegistration { + id: string; publicKey: Uint8Array; counter: number; transports: string | null; rpId: string; + aaguid: string | null; deviceType: string; backedUp: number; label: string | null; createdAt: number; +} + +export function setup(db: Database) { + return { + createPrincipal: (input: CreatePrincipalInput) => createPrincipal(db, input), + revokePrincipal: (id: string, now: number) => revokePrincipal(db, id, now), + revokeAll: (now: number) => revokeAllPrincipals(db, now), + createGrant: (input: Parameters[1]) => createReferenceGrant(db, input), + revokeGrant: (input: Parameters[1]) => revokeReferenceGrant(db, input), + passkeyCounter(id: string, rpId: string, counter: number, at: number) { + const row = db.query("SELECT rp_id FROM passkey_credentials WHERE id = ?").get(id) as { rp_id: string } | null; + if (!row || row.rp_id !== rpId) return false; + db.prepare("UPDATE passkey_credentials SET counter = ?, last_used_at = ? WHERE id = ?").run(counter, at, id); + return true; + }, + registerPasskey(input: PasskeyRegistration) { + db.prepare(`INSERT INTO passkey_credentials + (id, public_key, counter, transports, rp_id, aaguid, device_type, backed_up, label, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`) + .run(input.id, input.publicKey, input.counter, input.transports, input.rpId, input.aaguid, + input.deviceType, input.backedUp, input.label, input.createdAt); + }, + labelPasskey(id: string, label: string | null) { + return db.prepare("UPDATE passkey_credentials SET label = ? WHERE id = ?").run(label, id).changes; + }, + deletePasskey(id: string, at: number) { + const changes = db.prepare("DELETE FROM passkey_credentials WHERE id = ?").run(id).changes; + // Keep the original two autocommits and their order: delete first, then + // revoke the credential's sessions, before runtime invalidation/HTTP. + return { changes, revokedIds: changes ? revokeByCredential(db, id, at) : [] }; + }, + partitionSeed: () => accountPartitionSeed(db), + }; +} + +function revive(error: WireError): Error | undefined { + const as = (made: E) => Object.assign(made, { message: error.message }); + switch (error.name) { + case "PrincipalLimitError": return as(new PrincipalLimitError()); + case "PrincipalCredentialNotFoundError": return as(new PrincipalCredentialNotFoundError("")); + case "PrincipalCreatorNotUsableError": return as(new PrincipalCreatorNotUsableError(undefined)); + case "ReferenceRepositoryError": return new ReferenceRepositoryError(error.message); + case "ReferenceGrantPrincipalError": return as(new ReferenceGrantPrincipalError()); + case "ReferenceGrantDuplicateError": return as(new ReferenceGrantDuplicateError()); + case "ReferenceGrantLimitError": return as(new ReferenceGrantLimitError()); + case "TypeError": return new TypeError(error.message); + case "RangeError": return new RangeError(error.message); + default: return undefined; + } +} + +type Procedures = ReturnType; +type AccountWrites = { [K in keyof Procedures]: (...args: Parameters) => Promise> }; +const registered = new WeakMap(); + +/** Standalone callers use the same procedures inline; hosted callers register their shared writer. */ +export function accountWrites(db: Database): AccountWrites { + let writes = registered.get(db); + if (!writes) { writes = createAccountWrites(db); registered.set(db, writes); } + return writes; +} + +export function registerAccountWrites(db: Database, writer: SqliteWriter): AccountWrites { + const writes = createAccountWrites(db, writer); + registered.set(db, writes); + return writes; +} + +function createAccountWrites(db: Database, writer?: SqliteWriter): AccountWrites { + const call = procedureCaller(db, { url, setup }, undefined, writer, revive); + return { + createPrincipal: input => call("createPrincipal", input), + revokePrincipal: (id, now) => call("revokePrincipal", id, now), + revokeAll: now => call("revokeAll", now), + createGrant: input => call("createGrant", input), + revokeGrant: input => call("revokeGrant", input), + passkeyCounter: (id, rpId, counter, at) => call("passkeyCounter", id, rpId, counter, at), + registerPasskey: input => call("registerPasskey", input), + labelPasskey: (id, label) => call("labelPasskey", id, label), + deletePasskey: (id, at) => call("deletePasskey", id, at), + partitionSeed: () => call("partitionSeed"), + }; +} diff --git a/packages/ui-server/src/db/principal-writes.ts b/packages/ui-server/src/db/principal-writes.ts index cca990a2e9..8d5d844777 100644 --- a/packages/ui-server/src/db/principal-writes.ts +++ b/packages/ui-server/src/db/principal-writes.ts @@ -5,8 +5,8 @@ * login's principal on every few requests; each of those commits on the * `ui.db` write worker at `synchronous=FULL` instead of the app connection. * The procedures are the synchronous writers in `principals.ts`, run on the - * write connection. Revocations and administrative writes still commit on the - * app connection. + * write connection. Account administration uses `account-writes.ts` on the + * same worker. */ import type { Database } from "bun:sqlite"; diff --git a/packages/ui-server/src/middleware/account-partition.ts b/packages/ui-server/src/middleware/account-partition.ts index d97509aeca..de97db04a0 100644 --- a/packages/ui-server/src/middleware/account-partition.ts +++ b/packages/ui-server/src/middleware/account-partition.ts @@ -81,9 +81,9 @@ export function deriveAccountPartitionKey(seed: string, root: string, scope: str } /** The key for each request's principal, on one host: the seed is read once. */ -export function createAccountPartitionKeys(db: Database, brainPath: string, authMode: AuthMode) { +export function createAccountPartitionKeys(db: Database, brainPath: string, authMode: AuthMode, hostSeed?: string) { let root: string | null = null; - let seed: string | null = null; + let seed: string | null = hostSeed ?? null; return (principal: Principal | undefined): string | null => { if (!principal) return null; const scope = accountScope(principal, authMode); diff --git a/packages/ui-server/src/middleware/auth.ts b/packages/ui-server/src/middleware/auth.ts index 54384a57ff..ac1b9380f0 100644 --- a/packages/ui-server/src/middleware/auth.ts +++ b/packages/ui-server/src/middleware/auth.ts @@ -1,3 +1,4 @@ +import { accountWrites } from "../db/account-writes.js"; import type { Counter } from "@opentelemetry/api"; import type { Logger } from "@opentelemetry/api-logs"; import { Hono } from "hono"; @@ -14,8 +15,6 @@ import { PrincipalCredentialNotFoundError, PrincipalLimitError, resolvePrincipal, - revokePrincipal, - revokeAllPrincipals, type Principal, } from "../db/principals.js"; import { principalWrites } from "../db/principal-writes.js"; @@ -445,11 +444,11 @@ export function applyPrincipalRevocation( /** * Revoke every cookie-bearing principal and disconnect its attached clients. */ -export function revokeAllSessions( +export async function revokeAllSessions( db: Database, revoker: Pick -): void { - const revokedIds = revokeAllPrincipals(db, Date.now()); +): Promise { + const revokedIds = await accountWrites(db).revokeAll(Date.now()); applyPrincipalRevocation(revoker, revokedIds); } @@ -802,11 +801,11 @@ export function authRoutes( } c.set("principal", principal); if (principal.kind === "owner") { - revokeAllSessions(deps.db, deps.revoker); + await revokeAllSessions(deps.db, deps.revoker); } else { applyPrincipalRevocation( deps.revoker, - revokePrincipal(deps.db, principal.id, Date.now()) + await accountWrites(deps.db).revokePrincipal(principal.id, Date.now()) ); } } diff --git a/packages/ui-server/src/middleware/passkeys.ts b/packages/ui-server/src/middleware/passkeys.ts index 7eb0c13e8b..576fd19b44 100644 --- a/packages/ui-server/src/middleware/passkeys.ts +++ b/packages/ui-server/src/middleware/passkeys.ts @@ -29,7 +29,7 @@ import type { WebAuthnConfig } from "../config/env.js"; import { readJsonBody } from "./body-limit.js"; import { requireJson } from "./origin.js"; import type { AppEnv } from "../app-env.js"; -import { revokeByCredential } from "../db/principals.js"; +import { accountWrites } from "../db/account-writes.js"; import type { WsHost } from "../ws/host.js"; import { requireOwner } from "./require-owner.js"; @@ -409,14 +409,7 @@ export function passkeyPublicRoutes( }); return fail(); } - const currentRow = credentialById(ctx.db, row.id); - if (!currentRow || currentRow.rp_id !== rp.rpID) return fail(); - - ctx.db - .prepare( - "UPDATE passkey_credentials SET counter = ?, last_used_at = ? WHERE id = ?" - ) - .run(result.authenticationInfo.newCounter, now(), currentRow.id); + if (!await accountWrites(ctx.db).passkeyCounter(row.id, rp.rpID, result.authenticationInfo.newCounter, now())) return fail(); } finally { releasePasskeyVerification(ip); } @@ -541,26 +534,12 @@ export function passkeyManagementRoutes( } const { credential, aaguid, credentialDeviceType, credentialBackedUp } = result.registrationInfo; - const createdAt = now(); - ctx.db - .prepare( - `INSERT INTO passkey_credentials - (id, public_key, counter, transports, rp_id, aaguid, device_type, - backed_up, label, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` - ) - .run( - credential.id, - credential.publicKey, - credential.counter, - credential.transports ? JSON.stringify(credential.transports) : null, - rp.rpID, - aaguid || null, - credentialDeviceType, - credentialBackedUp ? 1 : 0, - sanitizeLabel(body.label), - createdAt - ); + await accountWrites(ctx.db).registerPasskey({ + id: credential.id, publicKey: credential.publicKey, counter: credential.counter, + transports: credential.transports ? JSON.stringify(credential.transports) : null, + rpId: rp.rpID, aaguid: aaguid || null, deviceType: credentialDeviceType, + backedUp: credentialBackedUp ? 1 : 0, label: sanitizeLabel(body.label), createdAt: now(), + }); const row = credentialById(ctx.db, credential.id); return c.json({ ok: true, credential: row ? toSummary(row) : null }); } catch (err) { @@ -598,24 +577,17 @@ export function passkeyManagementRoutes( } catch { return c.json({ error: "Invalid request body" }, 400); } - const result = ctx.db - .prepare("UPDATE passkey_credentials SET label = ? WHERE id = ?") - .run(sanitizeLabel(body.label), c.req.param("id")); - if (result.changes === 0) return c.json({ error: "Unknown passkey" }, 404); + const changes = await accountWrites(ctx.db).labelPasskey(c.req.param("id"), sanitizeLabel(body.label)); + if (changes === 0) return c.json({ error: "Unknown passkey" }, 404); return c.json({ ok: true }); }); - app.delete("/auth/passkey/:id", (c) => { + app.delete("/auth/passkey/:id", async (c) => { if (mode !== "password") return notEnabled(c); const credentialId = c.req.param("id"); - const result = ctx.db - .prepare("DELETE FROM passkey_credentials WHERE id = ?") - .run(credentialId); + const result = await accountWrites(ctx.db).deletePasskey(credentialId, Date.now()); if (result.changes === 0) return c.json({ error: "Unknown passkey" }, 404); - applyPrincipalRevocation( - ctx.revoker, - revokeByCredential(ctx.db, credentialId, Date.now()) - ); + applyPrincipalRevocation(ctx.revoker, result.revokedIds); return c.json({ ok: true }); }); diff --git a/packages/ui-server/src/middleware/principals.ts b/packages/ui-server/src/middleware/principals.ts index 8450b4a36f..5e79b6768d 100644 --- a/packages/ui-server/src/middleware/principals.ts +++ b/packages/ui-server/src/middleware/principals.ts @@ -6,25 +6,22 @@ import { generateSignedCookie } from "hono/cookie"; import type { AppEnv } from "../app-env.js"; import { - createPrincipal, isCookieBearingPrincipal, isUsablePrincipal, listLivePrincipals, PrincipalCreatorNotUsableError, PrincipalLimitError, resolvePrincipal, - revokePrincipal, } from "../db/principals.js"; import { - createReferenceGrant, listReferenceGrants, ReferenceGrantDuplicateError, ReferenceGrantLimitError, ReferenceGrantPrincipalError, ReferenceRepositoryError, - revokeReferenceGrant, type ReferenceGrant, } from "../db/reference-grants.js"; +import { accountWrites } from "../db/account-writes.js"; import { readJsonBody } from "./body-limit.js"; import { applyPrincipalRevocation, @@ -152,7 +149,7 @@ export function principalManagementRoutes( const caller = c.get("principal")!; let principal; try { - principal = createPrincipal(ctx.db, { + principal = await accountWrites(ctx.db).createPrincipal({ authMethod: "delegated", label, createdBy: caller.id, @@ -193,10 +190,10 @@ export function principalManagementRoutes( }); }); - app.delete("/auth/principals/:id", (c) => { + app.delete("/auth/principals/:id", async (c) => { const caller = c.get("principal")!; const targetId = c.req.param("id"); - const revokedIds = revokePrincipal(ctx.db, targetId, Date.now()); + const revokedIds = await accountWrites(ctx.db).revokePrincipal(targetId, Date.now()); if (revokedIds.length === 0) { return c.json({ error: "Unknown principal" }, 404); } @@ -243,7 +240,7 @@ export function principalManagementRoutes( const principalId = c.req.param("id"); let grant: ReferenceGrant; try { - grant = createReferenceGrant(ctx.db, { + grant = await accountWrites(ctx.db).createGrant({ brain: ctx.brain, principalId, repositoryPrefix: body.repositoryPrefix, @@ -272,10 +269,10 @@ export function principalManagementRoutes( return c.json(grantJson(grant), 201); }); - app.delete("/auth/principals/:id/reference-grants/:grantId", (c) => { + app.delete("/auth/principals/:id/reference-grants/:grantId", async (c) => { const caller = c.get("principal")!; const principalId = c.req.param("id"); - const revoked = revokeReferenceGrant(ctx.db, { + const revoked = await accountWrites(ctx.db).revokeGrant({ brain: ctx.brain, principalId, id: c.req.param("grantId"), diff --git a/packages/ui-server/src/routes/push.ts b/packages/ui-server/src/routes/push.ts index f26a095956..d2fc43edae 100644 --- a/packages/ui-server/src/routes/push.ts +++ b/packages/ui-server/src/routes/push.ts @@ -1,7 +1,7 @@ import { Hono } from "hono"; import { z } from "zod"; -import type { PushSender } from "../activity/push-sender.js"; +import type { AwaitablePushSender } from "../activity/push-sender.js"; import { noticeClientId } from "../inbox/notify.js"; import type { AwaitableActionNotifier } from "../inbox/notify-writes.js"; import type { AppEnv } from "../app-env.js"; @@ -39,13 +39,13 @@ const zoneSchema = z.object({ clientId: z.string().max(64).optional(), }); -export function createPushRoutes(deps: { sender: PushSender; notices?: AwaitableActionNotifier }) { +export function createPushRoutes(deps: { sender: AwaitablePushSender; notices?: AwaitableActionNotifier }) { const { sender, notices } = deps; return new Hono() - .get("/push/public-key", (c) => { + .get("/push/public-key", async (c) => { try { - return c.json({ publicKey: sender.publicKey() }); + return c.json({ publicKey: await sender.publicKey() }); } catch (err) { return c.json( { error: err instanceof Error ? err.message : "Push unavailable" }, @@ -82,7 +82,7 @@ export function createPushRoutes(deps: { sender: PushSender; notices?: Awaitable if (result instanceof Response) return result; const body = subscribeSchema.parse(result); const principal = c.get("principal")!; - if (!sender.subscribe(body.subscription, principal.id, body.label)) { + if (!await sender.subscribe(body.subscription, principal.id, body.label)) { return c.json({ error: "Authentication required" }, 401); } // Omitted: an older client keeps the device's last reported zone. @@ -122,7 +122,7 @@ export function createPushRoutes(deps: { sender: PushSender; notices?: Awaitable if (result instanceof Response) return result; const body = unsubscribeSchema.parse(result); return c.json({ - removed: sender.unsubscribe(body.endpoint, c.get("principal")!.id), + removed: await sender.unsubscribe(body.endpoint, c.get("principal")!.id), }); } catch (err) { return c.json( diff --git a/packages/ui-server/src/storage/sampling.ts b/packages/ui-server/src/storage/sampling.ts index 941e9fb4e7..7fa5751df3 100644 --- a/packages/ui-server/src/storage/sampling.ts +++ b/packages/ui-server/src/storage/sampling.ts @@ -7,7 +7,7 @@ import type { StorageNotifier } from "./notify.js"; import type { SharedStorageScan } from "./shared-scan.js"; import { bucket } from "./history-model.js"; /** Background work starts at the next UTC six-hour boundary, never at boot or on a read. */ -export function createStorageSampling(db: Database, storage: StorageCoordinator, history: StorageHistory, shared: SharedStorageScan, notices: StorageNotifier, sender: PushSender, bootAt = Date.now()) { +export function createStorageSampling(db: Database, storage: StorageCoordinator, history: StorageHistory, shared: SharedStorageScan, notices: StorageNotifier, sender: Pick, bootAt = Date.now()) { let lastBucket = bucket(0, bootAt), lastClock = bootAt, lastCleanup = bootAt; let delivering: Promise | null = null, cleaning: Promise | null = null; const controller = new AbortController(); diff --git a/packages/ui-server/src/ws/host.ts b/packages/ui-server/src/ws/host.ts index 8013735f6e..0d8b72dac8 100644 --- a/packages/ui-server/src/ws/host.ts +++ b/packages/ui-server/src/ws/host.ts @@ -11,7 +11,7 @@ import { createSilentObservability, type Observability } from "../observability/ import { FrameRateLimiter } from "./rate-limit.js"; import type { AwaitableActivityStore } from "../activity/store.js"; import type { ActivityStream } from "../activity/stream.js"; -import type { PushSender } from "../activity/push-sender.js"; +import type { AwaitablePushSender } from "../activity/push-sender.js"; import type { RuntimeStatus } from "../activity/runtime-status.js"; import type { GovernanceEventInput } from "../governance/events.js"; import type { Principal } from "../db/principals.js"; @@ -34,7 +34,7 @@ export interface ActivityRuntime { store: AwaitableActivityStore; stream: ActivityStream; /** Push bindings invalidated through the same principal-revocation boundary. */ - pushSender?: Pick; + pushSender?: Pick; /** Read seam for the agent-facing query tool (bridge.queryActivity). */ query?: (query: import("@schlessera/brain-ui-sdk/server").ActivityQuery) => Record; /** Where each turn's runtime report and auth failures are kept for /api/status. */ @@ -754,7 +754,9 @@ export class WsHost { this.clients.closeFor(principalId, code, reason); this.activity?.stream.dropFor(principalId); this.inbox?.dropFor(principalId); - this.activity?.pushSender?.unbindPrincipal(principalId); + const unbound = this.activity?.pushSender?.unbindPrincipal(principalId); + if (unbound instanceof Promise) void unbound.catch(error => this.log.emit({ severityText: "WARN", + body: "push principal unbind failed", attributes: { error: error instanceof Error ? error.message : String(error) } })); } for (const turn of affectedRunning) { turn.recorder?.recordPrincipalRevocation(turn.principalId); diff --git a/packages/ui-server/tests/account-writes-offloop.test.ts b/packages/ui-server/tests/account-writes-offloop.test.ts new file mode 100644 index 0000000000..163f01405c --- /dev/null +++ b/packages/ui-server/tests/account-writes-offloop.test.ts @@ -0,0 +1,273 @@ +/** Account/security administration commits on the shared FULL writer (#3002). */ +import { expect, test } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { Hono } from "hono"; +import { serializeSigned } from "hono/utils/cookie"; +import type { AppEnv } from "../src/app-env.js"; +import { createUiDb } from "../src/db/client.js"; +import { openSqliteWriter } from "../src/db/write-worker.js"; +import { createPrincipal, PrincipalCreatorNotUsableError, PrincipalCredentialNotFoundError, PrincipalLimitError, resolvePrincipal } from "../src/db/principals.js"; +import { accountWrites, registerAccountWrites } from "../src/db/account-writes.js"; +import { registerPrincipalWrites } from "../src/db/principal-writes.js"; +import { createReferenceGrant, ReferenceGrantDuplicateError, ReferenceGrantPrincipalError, ReferenceRepositoryError } from "../src/db/reference-grants.js"; +import { principalManagementRoutes } from "../src/middleware/principals.js"; +import { authGuard, authRoutes, createLoginRateLimiter, type AuthRuntime } from "../src/middleware/auth.js"; +import { createPasskeyChallenges, passkeyManagementRoutes, passkeyPublicRoutes, type PasskeyContext, type PasskeyDeps } from "../src/middleware/passkeys.js"; +import { createAccountPartitionKeys } from "../src/middleware/account-partition.js"; +import { resolveServerConfig } from "../src/config/env.js"; +import { createPushSender } from "../src/activity/push-sender.js"; +import { createPushRoutes } from "../src/routes/push.js"; +import { watchWriteSites } from "./helpers/write-sites.js"; +import * as probe from "./fixtures/write-worker-probe.js"; +import { expectRejection } from "./helpers/expect-rejection.js"; +import type { ActivityNotifier } from "../src/activity/notify.js"; + +const config = resolveServerConfig({ AUTH_MODE: "password", BRAIN_UI_PASSWORD_HASH: "odysseus-password-hash", + COOKIE_SECRET: "odysseus-cookie-secret-0123456789abcdef" }); +const auth: AuthRuntime = { ...config.auth, host: config.host }; +const revoker = { revokePrincipals() {} }; + +function fixture() { + const root = mkdtempSync(join(tmpdir(), "account-writes-odysseus-")); + const db = createUiDb(join(root, "ui.db")), writer = openSqliteWriter(db); + const owner = createPrincipal(db, { authMethod: "password", label: "Penelope", ttlSeconds: 3600 }); + const agent = createPrincipal(db, { authMethod: "delegated", label: "Athena", createdBy: owner.id, ttlSeconds: 3600 }); + registerPrincipalWrites(db, writer); + registerAccountWrites(db, writer); + const app = new Hono().use("*", async (c, next) => { c.set("principal", owner); await next(); }); + app.onError((error, c) => c.json({ error: error.message }, 500)); + app.route("/api", principalManagementRoutes("password", auth, { db, revoker, brain: "ithaca" })); + return { root, db, writer, owner, agent, app, + async close() { await writer.close(); db.close(); rmSync(root, { recursive: true, force: true }); } }; +} + +function json(method: string, body?: unknown): RequestInit { + return { method, headers: { "content-type": "application/json" }, ...(body === undefined ? {} : { body: JSON.stringify(body) }) }; +} + +for (const operation of ["mint", "revoke", "grant", "revokeGrant"] as const) { + test(`account ${operation} commits on the write worker at FULL, never on the app connection`, async () => { + const f = fixture(); + try { + const grant = operation === "revokeGrant" ? createReferenceGrant(f.db, { brain: "ithaca", principalId: f.agent.id, + repositoryPrefix: "https://github.com/odysseus", credential: null, createdBy: f.owner.id, now: Date.now() }) : null; + const sites = await watchWriteSites(f.db, f.writer, ["principals", "reference_grants"]); + expect(f.writer.offloop).toBe(true); + const response = operation === "mint" ? await f.app.request("/api/auth/principals", json("POST", { label: "Telemachus" })) + : operation === "revoke" ? await f.app.request(`/api/auth/principals/${f.agent.id}`, json("DELETE")) + : operation === "grant" ? await f.app.request(`/api/auth/principals/${f.agent.id}/reference-grants`, json("POST", { repositoryPrefix: "https://github.com/odysseus" })) + : await f.app.request(`/api/auth/principals/${f.agent.id}/reference-grants/${grant!.id}`, json("DELETE")); + const body = await response.json(); + expect({ status: response.status, body }, `ACCOUNT_${operation.toUpperCase()}_COMMITTED_BEFORE_HTTP_RESPONSE`) + .toMatchObject({ status: operation === "grant" ? 201 : 200 }); + expect(await sites.log(), `ACCOUNT_${operation.toUpperCase()}_WRITES_ON_WORKER`) + .toContain(operation === "grant" || operation === "revokeGrant" ? "reference_grants" : "principals"); + if (operation === "revoke") expect(resolvePrincipal(f.db, f.agent.id)?.revokedAt, "ACCOUNT_REVOCATION_COMMITTED").not.toBeNull(); + } finally { await f.close(); } + }); +} + +for (const operation of ["publicKey", "subscribe", "unsubscribe", "unbindPrincipal"] as const) { + test(`push ${operation} commits on the write worker at FULL, never on the app connection`, async () => { + const f = fixture(); + try { + const sender = createPushSender(f.db, { writer: f.writer }); + const sub = { endpoint: "https://push.example.test/ithaca", keys: { p256dh: "bow", auth: "loom" } }; + if (operation === "unsubscribe" || operation === "unbindPrincipal") await sender.subscribe(sub, f.owner.id); + const sites = await watchWriteSites(f.db, f.writer, ["push_subscriptions", "vapid_keys"]); + const result = operation === "publicKey" ? await sender.publicKey() + : operation === "subscribe" ? await sender.subscribe(sub, f.owner.id) + : operation === "unsubscribe" ? await sender.unsubscribe(sub.endpoint, f.owner.id) : await sender.unbindPrincipal(f.owner.id); + expect(result, `PUSH_${operation.toUpperCase()}_COMMITTED`).toBeTruthy(); + expect(await sites.log(), `PUSH_${operation.toUpperCase()}_WRITES_ON_WORKER`) + .toEqual([operation === "publicKey" ? "vapid_keys" : "push_subscriptions"]); + } finally { await f.close(); } + }); +} + +test("a revocation route waits behind the worker and every authentication after its answer refuses the principal", async () => { + const f = fixture(); + try { + const cookie = await serializeSigned("brain_ui_session", f.agent.id, auth.cookieSecret!); + const app = new Hono().use("/api/*", authGuard("password", auth, f.db)) + .get("/api/who", c => c.json({ id: c.get("principal")!.id })); + const held = f.writer.channel(probe, { table: "odysseus_log" }); await held.ready; + const slow = held.call("slow", "Calypso", 250); + let answered = false; + const response = Promise.resolve(f.app.request(`/api/auth/principals/${f.agent.id}`, json("DELETE"))).then(r => { answered = true; return r; }); + // Let the asynchronous HTTP handler submit while the worker is held. + await new Promise(resolve => setTimeout(resolve, 50)); + expect(answered, "ACCOUNT_REVOCATION_HTTP_WAITS_FOR_COMMIT").toBe(false); + await slow; + expect((await response).status, "ACCOUNT_REVOCATION_HTTP_SUCCEEDS").toBe(200); + for (let i = 0; i < 3; i++) expect((await app.request("/api/who", { headers: { cookie } })).status, + "ACCOUNT_REVOKED_PRINCIPAL_NEVER_AUTHENTICATES_AFTER_RESPONSE").toBe(401); + } finally { await f.close(); } +}); + +test("push HTTP routes await subscription commits and key creation", async () => { + const f = fixture(); + try { + const sender = createPushSender(f.db, { writer: f.writer }); + f.app.route("/api", createPushRoutes({ sender })); + const sites = await watchWriteSites(f.db, f.writer, ["push_subscriptions", "vapid_keys"]); + const key = await f.app.request("/api/push/public-key"); + expect(await key.json(), "PUSH_KEY_HTTP_AWAITS_COMMIT").toMatchObject({ publicKey: expect.any(String) }); + const sub = { endpoint: "https://push.example.test/ithaca", keys: { p256dh: "bow", auth: "loom" } }; + expect((await f.app.request("/api/push/subscribe", json("POST", { subscription: sub }))).status).toBe(200); + expect(sender.subscriptions(f.owner.id), "PUSH_SUBSCRIBE_HTTP_AWAITS_COMMIT").toHaveLength(1); + expect(await (await f.app.request("/api/push/unsubscribe", json("POST", { endpoint: sub.endpoint }))).json(), + "PUSH_UNSUBSCRIBE_HTTP_AWAITS_COMMIT").toEqual({ removed: true }); + expect(await sites.log(), "PUSH_HTTP_WRITES_ON_WORKER").toEqual(["vapid_keys", "push_subscriptions", "push_subscriptions"]); + } finally { await f.close(); } +}); + +function passkeys(f: ReturnType) { + const ctx: PasskeyContext = { db: f.db, auth, revoker, rateLimiter: createLoginRateLimiter(), challenges: createPasskeyChallenges(), + webauthn: config.webauthn, allowedOrigins: ["http://localhost"] }; + const deps: PasskeyDeps = { + verifyRegistrationResponse: (async () => ({ verified: true, registrationInfo: { + fmt: "none", credentialType: "public-key", attestationObject: new Uint8Array(), userVerified: true, origin: "http://localhost", + aaguid: "ithaca", credential: { id: "odysseus-bow", publicKey: new Uint8Array([1, 2, 3]), counter: 0, transports: ["internal"] }, + credentialDeviceType: "multiDevice", credentialBackedUp: true, + } })), + verifyAuthenticationResponse: (async () => ({ verified: true, authenticationInfo: { credentialID: "odysseus-bow", newCounter: 7, userVerified: true, + credentialDeviceType: "multiDevice", credentialBackedUp: true, origin: "http://localhost", rpID: "localhost" } })), + }; + f.app.route("/api", passkeyPublicRoutes("password", ctx, deps)); + f.app.route("/api", passkeyManagementRoutes("password", ctx, deps)); + const ceremony = (type: "registration" | "authentication") => { + const challenge = `ithaca-${type}`; + ctx.challenges.putChallenge(challenge, type, Date.now()); + return { id: "odysseus-bow", response: { clientDataJSON: Buffer.from(JSON.stringify({ challenge })).toString("base64url") } }; + }; + const request = (path: string, method: string, body?: unknown) => f.app.request(path, { ...json(method, body), + headers: { "content-type": "application/json", host: "localhost", origin: "http://localhost" } }); + const register = () => request("/api/auth/passkey/register-verify", "POST", { response: ceremony("registration"), label: "Odysseus' bow" }); + return { ceremony, request, register }; +} + +for (const operation of ["register", "counter", "label", "delete"] as const) { + test(`passkey ${operation} commits on the write worker at FULL, never on the app connection`, async () => { + const f = fixture(); + try { + const keys = passkeys(f); + if (operation !== "register") expect((await keys.register()).status, "PASSKEY_FIXTURE_REGISTERED").toBe(200); + const principal = operation === "delete" ? await accountWrites(f.db).createPrincipal({ authMethod: "passkey", label: "Odysseus", + credentialId: "odysseus-bow", ttlSeconds: 3600 }) : null; + const sites = await watchWriteSites(f.db, f.writer, ["passkey_credentials", "principals"]); + const response = operation === "register" ? await keys.register() : operation === "counter" + ? await keys.request("/api/auth/passkey/login-verify", "POST", keys.ceremony("authentication")) + : await keys.request("/api/auth/passkey/odysseus-bow", operation === "delete" ? "DELETE" : "PUT", { label: "Ithaca" }); + const body = await response.json(); + expect({ status: response.status, body }, `PASSKEY_${operation.toUpperCase()}_HTTP_AWAITS_COMMIT`).toMatchObject({ status: 200 }); + const row = f.db.query("SELECT counter, label FROM passkey_credentials WHERE id = 'odysseus-bow'").get(); + expect(row, `PASSKEY_${operation.toUpperCase()}_READ_SEES_COMMIT`).toEqual(operation === "delete" ? null : { + counter: operation === "counter" ? 7 : 0, label: operation === "label" ? "Ithaca" : "Odysseus' bow", + }); + const log = await sites.log(); + expect(log[0], `PASSKEY_${operation.toUpperCase()}_WRITES_ON_WORKER`).toBe("passkey_credentials"); + if (operation === "delete") { + expect(log, "PASSKEY_DELETE_BEFORE_SESSION_REVOCATION").toEqual(["passkey_credentials", "principals"]); + expect(resolvePrincipal(f.db, principal!.id)?.revokedAt, "PASSKEY_SESSION_REVOKED_BEFORE_RESPONSE").not.toBeNull(); + } + } finally { await f.close(); } + }); +} + +for (const who of ["owner", "agent"] as const) { + test(`${who} logout awaits its revocation on the FULL worker`, async () => { + const f = fixture(); + try { + f.app.route("/api", authRoutes("password", auth, { db: f.db, revoker, rateLimiter: createLoginRateLimiter() })); + const sites = await watchWriteSites(f.db, f.writer, ["principals"]); + const cookie = await serializeSigned("brain_ui_session", f[who].id, auth.cookieSecret!); + const response = await f.app.request("/api/auth/logout", { method: "POST", headers: { cookie } }); + expect(response.status, `ACCOUNT_LOGOUT_${who.toUpperCase()}_HTTP_AWAITS_COMMIT`).toBe(200); + expect(resolvePrincipal(f.db, f[who].id)?.revokedAt, "ACCOUNT_LOGOUT_DURABLE_REVOCATION").not.toBeNull(); + expect(await sites.log(), "ACCOUNT_LOGOUT_WRITES_ON_WORKER").toContain("principals"); + if (who === "owner") expect(resolvePrincipal(f.db, f.agent.id)?.revokedAt, "ACCOUNT_OWNER_LOGOUT_REVOKES_ALL").not.toBeNull(); + } finally { await f.close(); } + }); +} + +test("the partition seed commits on the FULL worker and concurrent first uses share it", async () => { + const f = fixture(); + try { + const sites = await watchWriteSites(f.db, f.writer, ["settings"]); + const writes = accountWrites(f.db); + const [first, raced] = await Promise.all([writes.partitionSeed(), writes.partitionSeed()]); + expect(first, "ACCOUNT_PARTITION_SEED_FORMAT").toMatch(/^[0-9a-f]{64}$/); + expect(first, "ACCOUNT_PARTITION_SEED_RACE_USES_COMMIT").toBe(raced); + const key = createAccountPartitionKeys(f.db, f.root, "password", first); + expect(key(f.owner), "ACCOUNT_PARTITION_KEY_READS_WORKER_SEED").toBe(createAccountPartitionKeys(f.db, f.root, "password")(f.owner)); + expect(await sites.log(), "ACCOUNT_PARTITION_SEED_INSERTED_ONCE_ON_WORKER").toEqual(["settings"]); + } finally { await f.close(); } +}); + +test("two delivery passes racing the key route create VAPID once and all use the committed pair", async () => { + const f = fixture(); + try { + const sent: Array<{ publicKey: string; privateKey: string }> = []; + const sender = createPushSender(f.db, { writer: f.writer, send: async (_sub, _body, options) => { sent.push(options.vapidDetails); } }); + await sender.subscribe({ endpoint: "https://push.example.test/ithaca", keys: { p256dh: "bow", auth: "loom" } }, f.owner.id); + const sites = await watchWriteSites(f.db, f.writer, ["vapid_keys"]); + const notifier = { pending: () => [{ id: 1, runId: "ithaca", title: "Ithaca", body: "Ready" }], markDelivered: async () => {} } as unknown as ActivityNotifier; + f.app.route("/api", createPushRoutes({ sender })); + const [one, two, response] = await Promise.all([sender.deliverPending(notifier), sender.deliverPending(notifier), f.app.request("/api/push/public-key")]); + const key = await response.json() as { publicKey: string }; + expect([one, two], "PUSH_RACING_PASSES_BOTH_SEND").toEqual([1, 1]); + expect(sent.map(pair => pair.publicKey), "PUSH_VAPID_RACE_SHARES_PUBLIC_KEY").toEqual([key.publicKey, key.publicKey]); + expect(sent[0]?.privateKey, "PUSH_VAPID_RACE_SHARES_PRIVATE_KEY").toBe(sent[1]?.privateKey); + expect(await sites.log(), "PUSH_VAPID_RACE_INSERTS_ONCE").toEqual(["vapid_keys"]); + } finally { await f.close(); } +}); + +test("account worker refusals retain their classes and reference-grant HTTP statuses", async () => { + const f = fixture(); + try { + const writes = accountWrites(f.db); + const input = { brain: "ithaca", principalId: f.agent.id, repositoryPrefix: "https://github.com/odysseus", credential: null, createdBy: f.owner.id, now: Date.now() }; + await writes.createGrant(input); + await expectRejection(writes.createGrant(input), ReferenceGrantDuplicateError, "ACCOUNT_GRANT_DUPLICATE_CLASS"); + await expectRejection(writes.createGrant({ ...input, principalId: "missing" }), ReferenceGrantPrincipalError, "ACCOUNT_GRANT_PRINCIPAL_CLASS"); + await expectRejection(writes.createGrant({ ...input, repositoryPrefix: "invalid" }), ReferenceRepositoryError, "ACCOUNT_GRANT_REPOSITORY_CLASS"); + for (const [id, repositoryPrefix, status] of [[f.agent.id, input.repositoryPrefix, 409], ["missing", input.repositoryPrefix, 404], [f.agent.id, "invalid", 400]] as const) { + expect((await f.app.request(`/api/auth/principals/${id}/reference-grants`, json("POST", { repositoryPrefix }))).status, + "ACCOUNT_GRANT_HTTP_REFUSAL_STATUS").toBe(status); + } + } finally { await f.close(); } +}); + + +test("queued administration rechecks owner authority and preserves principal refusal classes and statuses", async () => { + const f = fixture(); + try { + const writes = accountWrites(f.db); + await expectRejection(writes.createPrincipal({ authMethod: "passkey", label: "Odysseus", credentialId: "missing", ttlSeconds: 3600 }), + PrincipalCredentialNotFoundError, "ACCOUNT_CREDENTIAL_REFUSAL_CLASS"); + const held = f.writer.channel(probe, { table: "odysseus_log" }); await held.ready; + const slow = held.call("slow", "Calypso", 150); + const revocation = writes.revokePrincipal(f.owner.id, Date.now()); + const minted = writes.createPrincipal({ authMethod: "delegated", label: "Telemachus", createdBy: f.owner.id, ttlSeconds: 3600 }); + await expectRejection(minted, PrincipalCreatorNotUsableError, "ACCOUNT_CREATOR_REFUSAL_CLASS"); + await slow; await revocation; + expect((await f.app.request("/api/auth/principals", json("POST", { label: "Telemachus" }))).status, "ACCOUNT_CREATOR_REFUSAL_HTTP_STATUS").toBe(403); + expect((await f.app.request(`/api/auth/principals/${f.agent.id}/reference-grants`, json("POST", { repositoryPrefix: "https://github.com/odysseus" }))).status, + "ACCOUNT_GRANT_RECHECKS_REVOKED_CREATOR").toBe(404); + } finally { await f.close(); } +}); + +test("principal capacity refusal keeps its class and 503 across the account worker", async () => { + const f = fixture(); + try { + f.db.transaction(() => { + for (let i = 0; i < 98; i++) createPrincipal(f.db, { authMethod: "password", label: "Ithaca", ttlSeconds: 3600 }); + }).immediate(); + await expectRejection(accountWrites(f.db).createPrincipal({ authMethod: "password", label: "Odysseus", ttlSeconds: 3600 }), + PrincipalLimitError, "ACCOUNT_CAPACITY_REFUSAL_CLASS"); + expect((await f.app.request("/api/auth/principals", json("POST", { label: "Telemachus" }))).status, "ACCOUNT_CAPACITY_REFUSAL_HTTP_STATUS").toBe(503); + } finally { await f.close(); } +}); From b520a61b5d762ed4503683e1934dca8d8ed08710 Mon Sep 17 00:00:00 2001 From: Alain Schlesser Date: Mon, 26 Oct 2026 16:03:31 +0100 Subject: [PATCH 2/4] test(ui-server): retain worker refusal evidence in route assertions --- packages/ui-server/tests/account-writes-offloop.test.ts | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/packages/ui-server/tests/account-writes-offloop.test.ts b/packages/ui-server/tests/account-writes-offloop.test.ts index 163f01405c..0b56f83e27 100644 --- a/packages/ui-server/tests/account-writes-offloop.test.ts +++ b/packages/ui-server/tests/account-writes-offloop.test.ts @@ -125,7 +125,8 @@ test("push HTTP routes await subscription commits and key creation", async () => }); function passkeys(f: ReturnType) { - const ctx: PasskeyContext = { db: f.db, auth, revoker, rateLimiter: createLoginRateLimiter(), challenges: createPasskeyChallenges(), + const failures: string[] = []; + const ctx: PasskeyContext = { log: { emit: record => { failures.push(String(record.attributes?.error ?? record.body)); } }, db: f.db, auth, revoker, rateLimiter: createLoginRateLimiter(), challenges: createPasskeyChallenges(), webauthn: config.webauthn, allowedOrigins: ["http://localhost"] }; const deps: PasskeyDeps = { verifyRegistrationResponse: (async () => ({ verified: true, registrationInfo: { @@ -146,7 +147,7 @@ function passkeys(f: ReturnType) { const request = (path: string, method: string, body?: unknown) => f.app.request(path, { ...json(method, body), headers: { "content-type": "application/json", host: "localhost", origin: "http://localhost" } }); const register = () => request("/api/auth/passkey/register-verify", "POST", { response: ceremony("registration"), label: "Odysseus' bow" }); - return { ceremony, request, register }; + return { ceremony, request, register, failures }; } for (const operation of ["register", "counter", "label", "delete"] as const) { @@ -162,7 +163,7 @@ for (const operation of ["register", "counter", "label", "delete"] as const) { ? await keys.request("/api/auth/passkey/login-verify", "POST", keys.ceremony("authentication")) : await keys.request("/api/auth/passkey/odysseus-bow", operation === "delete" ? "DELETE" : "PUT", { label: "Ithaca" }); const body = await response.json(); - expect({ status: response.status, body }, `PASSKEY_${operation.toUpperCase()}_HTTP_AWAITS_COMMIT`).toMatchObject({ status: 200 }); + expect({ status: response.status, body, failures: keys.failures }, `PASSKEY_${operation.toUpperCase()}_HTTP_AWAITS_COMMIT`).toMatchObject({ status: 200 }); const row = f.db.query("SELECT counter, label FROM passkey_credentials WHERE id = 'odysseus-bow'").get(); expect(row, `PASSKEY_${operation.toUpperCase()}_READ_SEES_COMMIT`).toEqual(operation === "delete" ? null : { counter: operation === "counter" ? 7 : 0, label: operation === "label" ? "Ithaca" : "Odysseus' bow", @@ -185,7 +186,7 @@ for (const who of ["owner", "agent"] as const) { const sites = await watchWriteSites(f.db, f.writer, ["principals"]); const cookie = await serializeSigned("brain_ui_session", f[who].id, auth.cookieSecret!); const response = await f.app.request("/api/auth/logout", { method: "POST", headers: { cookie } }); - expect(response.status, `ACCOUNT_LOGOUT_${who.toUpperCase()}_HTTP_AWAITS_COMMIT`).toBe(200); + expect({ status: response.status, body: await response.json() }, `ACCOUNT_LOGOUT_${who.toUpperCase()}_HTTP_AWAITS_COMMIT`).toMatchObject({ status: 200 }); expect(resolvePrincipal(f.db, f[who].id)?.revokedAt, "ACCOUNT_LOGOUT_DURABLE_REVOCATION").not.toBeNull(); expect(await sites.log(), "ACCOUNT_LOGOUT_WRITES_ON_WORKER").toContain("principals"); if (who === "owner") expect(resolvePrincipal(f.db, f.agent.id)?.revokedAt, "ACCOUNT_OWNER_LOGOUT_REVOKES_ALL").not.toBeNull(); From 48e8af6a65194a4dbec8742e2aa0cc33a85ef313 Mon Sep 17 00:00:00 2001 From: Alain Schlesser Date: Mon, 26 Oct 2026 16:08:00 +0100 Subject: [PATCH 3/4] fix(ui-server): await push unbinding before revocation responses --- docs/decisions/sqlite-write-worker.md | 7 ++-- packages/ui-server/src/middleware/auth.ts | 8 ++--- packages/ui-server/src/middleware/passkeys.ts | 2 +- .../ui-server/src/middleware/principals.ts | 2 +- packages/ui-server/src/ws/host.ts | 11 +++--- .../tests/account-writes-offloop.test.ts | 34 +++++++++++++++++-- 6 files changed, 49 insertions(+), 15 deletions(-) diff --git a/docs/decisions/sqlite-write-worker.md b/docs/decisions/sqlite-write-worker.md index d95b564cdc..b115600f3c 100644 --- a/docs/decisions/sqlite-write-worker.md +++ b/docs/decisions/sqlite-write-worker.md @@ -190,9 +190,10 @@ are in [the governed-workers decision](governed-writable-workers.md#admitted-obj labels and deletion, reference grants and the account partition seed commit on the worker (`packages/ui-server/src/db/account-writes.ts`). The existing transaction bodies and modes stay together; passkey deletion still commits - before its credential's principal revocation. Every route awaits its writes - before answering or invalidating runtime authority, so every authentication - after a revocation response sees the committed revocation. Delegated minting + before its credential's principal revocation. Revocation routes await both + durable revocation and push unbinding before answering; client authority is + invalidated as soon as revocation commits. Every authentication after the + response sees the committed revocation. Delegated minting and reference grants recheck authority inside their original transactions; refusals keep their classes and HTTP statuses. Boot awaits the partition seed before constructing its synchronous key reader. Push subscription, removal diff --git a/packages/ui-server/src/middleware/auth.ts b/packages/ui-server/src/middleware/auth.ts index ac1b9380f0..4e86e943db 100644 --- a/packages/ui-server/src/middleware/auth.ts +++ b/packages/ui-server/src/middleware/auth.ts @@ -433,8 +433,8 @@ export async function resolveCookiePrincipal( export function applyPrincipalRevocation( revoker: Pick, principalIds: readonly string[] -): void { - revoker.revokePrincipals( +): void | Promise { + return revoker.revokePrincipals( principalIds, SESSION_REVOKED_CLOSE_CODE, SESSION_REVOKED_CLOSE_REASON @@ -449,7 +449,7 @@ export async function revokeAllSessions( revoker: Pick ): Promise { const revokedIds = await accountWrites(db).revokeAll(Date.now()); - applyPrincipalRevocation(revoker, revokedIds); + await applyPrincipalRevocation(revoker, revokedIds); } // --- proxy mode helpers --- @@ -803,7 +803,7 @@ export function authRoutes( if (principal.kind === "owner") { await revokeAllSessions(deps.db, deps.revoker); } else { - applyPrincipalRevocation( + await applyPrincipalRevocation( deps.revoker, await accountWrites(deps.db).revokePrincipal(principal.id, Date.now()) ); diff --git a/packages/ui-server/src/middleware/passkeys.ts b/packages/ui-server/src/middleware/passkeys.ts index 576fd19b44..fd29bfe355 100644 --- a/packages/ui-server/src/middleware/passkeys.ts +++ b/packages/ui-server/src/middleware/passkeys.ts @@ -587,7 +587,7 @@ export function passkeyManagementRoutes( const credentialId = c.req.param("id"); const result = await accountWrites(ctx.db).deletePasskey(credentialId, Date.now()); if (result.changes === 0) return c.json({ error: "Unknown passkey" }, 404); - applyPrincipalRevocation(ctx.revoker, result.revokedIds); + await applyPrincipalRevocation(ctx.revoker, result.revokedIds); return c.json({ ok: true }); }); diff --git a/packages/ui-server/src/middleware/principals.ts b/packages/ui-server/src/middleware/principals.ts index 5e79b6768d..09e1669ed7 100644 --- a/packages/ui-server/src/middleware/principals.ts +++ b/packages/ui-server/src/middleware/principals.ts @@ -198,7 +198,7 @@ export function principalManagementRoutes( return c.json({ error: "Unknown principal" }, 404); } - applyPrincipalRevocation(ctx.revoker, revokedIds); + await applyPrincipalRevocation(ctx.revoker, revokedIds); ctx.referenceAuthorityRevoked?.({ grantIds: [], principalIds: revokedIds }); ctx.log?.emit({ severityText: "INFO", diff --git a/packages/ui-server/src/ws/host.ts b/packages/ui-server/src/ws/host.ts index 0d8b72dac8..ef94f3800b 100644 --- a/packages/ui-server/src/ws/host.ts +++ b/packages/ui-server/src/ws/host.ts @@ -746,21 +746,22 @@ export class WsHost { * Running turns are marked but deliberately not aborted. * @internal Host wiring (#1053). */ - revokePrincipals(principalIds: readonly string[], code: number, reason: string): void { + revokePrincipals(principalIds: readonly string[], code: number, reason: string): void | Promise { const revoked = new Set(principalIds); if (revoked.size === 0) return; const affectedRunning = this.coordinator.revokePrincipals(revoked); + const unbinding: Promise[] = []; for (const principalId of revoked) { this.clients.closeFor(principalId, code, reason); this.activity?.stream.dropFor(principalId); this.inbox?.dropFor(principalId); const unbound = this.activity?.pushSender?.unbindPrincipal(principalId); - if (unbound instanceof Promise) void unbound.catch(error => this.log.emit({ severityText: "WARN", - body: "push principal unbind failed", attributes: { error: error instanceof Error ? error.message : String(error) } })); + if (unbound instanceof Promise) unbinding.push(unbound); } for (const turn of affectedRunning) { turn.recorder?.recordPrincipalRevocation(turn.principalId); } + if (unbinding.length) return Promise.all(unbinding).then(() => {}); } /** @@ -770,11 +771,13 @@ export class WsHost { expireAuthorizationContexts(now: number = Date.now()): void { const expired = new Set(); this.coordinator.collectExpiredPrincipalIds(now, expired); - this.revokePrincipals( + const revoked = this.revokePrincipals( [...expired], SESSION_EXPIRED_CLOSE_CODE, SESSION_EXPIRED_CLOSE_REASON ); + if (revoked instanceof Promise) void revoked.catch(error => this.log.emit({ severityText: "WARN", + body: "expired principal push unbind failed", attributes: { error: error instanceof Error ? error.message : String(error) } })); } /** Stop host-owned timers during application/test teardown. */ diff --git a/packages/ui-server/tests/account-writes-offloop.test.ts b/packages/ui-server/tests/account-writes-offloop.test.ts index 0b56f83e27..9d311e213c 100644 --- a/packages/ui-server/tests/account-writes-offloop.test.ts +++ b/packages/ui-server/tests/account-writes-offloop.test.ts @@ -23,13 +23,14 @@ import { watchWriteSites } from "./helpers/write-sites.js"; import * as probe from "./fixtures/write-worker-probe.js"; import { expectRejection } from "./helpers/expect-rejection.js"; import type { ActivityNotifier } from "../src/activity/notify.js"; +import { WsHost } from "../src/ws/host.js"; const config = resolveServerConfig({ AUTH_MODE: "password", BRAIN_UI_PASSWORD_HASH: "odysseus-password-hash", COOKIE_SECRET: "odysseus-cookie-secret-0123456789abcdef" }); const auth: AuthRuntime = { ...config.auth, host: config.host }; const revoker = { revokePrincipals() {} }; -function fixture() { +function fixture(revokerOverride: Pick = revoker) { const root = mkdtempSync(join(tmpdir(), "account-writes-odysseus-")); const db = createUiDb(join(root, "ui.db")), writer = openSqliteWriter(db); const owner = createPrincipal(db, { authMethod: "password", label: "Penelope", ttlSeconds: 3600 }); @@ -38,7 +39,7 @@ function fixture() { registerAccountWrites(db, writer); const app = new Hono().use("*", async (c, next) => { c.set("principal", owner); await next(); }); app.onError((error, c) => c.json({ error: error.message }, 500)); - app.route("/api", principalManagementRoutes("password", auth, { db, revoker, brain: "ithaca" })); + app.route("/api", principalManagementRoutes("password", auth, { db, revoker: revokerOverride, brain: "ithaca" })); return { root, db, writer, owner, agent, app, async close() { await writer.close(); db.close(); rmSync(root, { recursive: true, force: true }); } }; } @@ -272,3 +273,32 @@ test("principal capacity refusal keeps its class and 503 across the account work expect((await f.app.request("/api/auth/principals", json("POST", { label: "Telemachus" }))).status, "ACCOUNT_CAPACITY_REFUSAL_HTTP_STATUS").toBe(503); } finally { await f.close(); } }); + +test("revocation disconnects immediately and HTTP awaits the host's push unbinding commit", async () => { + let submitted!: () => void, release!: () => void; + const submission = new Promise(resolve => { submitted = resolve; }); + const commit = new Promise(resolve => { release = () => resolve(1); }); + const closed: string[] = []; + const host = { coordinator: { revokePrincipals: () => [] }, clients: { closeFor: (id: string) => closed.push(id) }, + activity: { stream: { dropFor() {} }, pushSender: { unbindPrincipal: () => { submitted(); return commit; } } } }; + const f = fixture({ revokePrincipals: (ids, code, reason) => WsHost.prototype.revokePrincipals.call(host as unknown as WsHost, ids, code, reason) }); + try { + let answered = false; + const response = Promise.resolve(f.app.request(`/api/auth/principals/${f.agent.id}`, json("DELETE"))).then(r => { answered = true; return r; }); + await submission; + expect(closed, "ACCOUNT_REVOCATION_DISCONNECTS_BEFORE_UNBIND_COMMIT").toEqual([f.agent.id]); + expect(answered, "ACCOUNT_REVOCATION_HTTP_AWAITS_PUSH_UNBIND_COMMIT").toBe(false); + release(); + expect((await response).status, "ACCOUNT_REVOCATION_HTTP_FOLLOWS_UNBIND_COMMIT").toBe(200); + } finally { release(); await f.close(); } +}); + +test("a failed push unbind still fails the revocation HTTP response after durable revocation", async () => { + const f = fixture({ revokePrincipals: async () => { throw new Error("Ithaca push unbind refused"); } }); + try { + const response = await f.app.request(`/api/auth/principals/${f.agent.id}`, json("DELETE")); + expect(response.status, "ACCOUNT_UNBIND_REFUSAL_HTTP_STATUS").toBe(500); + expect(await response.json(), "ACCOUNT_UNBIND_REFUSAL_MESSAGE").toEqual({ error: "Ithaca push unbind refused" }); + expect(resolvePrincipal(f.db, f.agent.id)?.revokedAt, "ACCOUNT_UNBIND_REFUSAL_KEEPS_DURABLE_REVOCATION").not.toBeNull(); + } finally { await f.close(); } +}); From 21f820efc6ebac1fad92082db8cd8a636ebd2c6b Mon Sep 17 00:00:00 2001 From: Alain Schlesser Date: Mon, 26 Oct 2026 16:11:39 +0100 Subject: [PATCH 4/4] test(ui-server): verify push unbind wait across the HTTP turn --- packages/ui-server/tests/account-writes-offloop.test.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/packages/ui-server/tests/account-writes-offloop.test.ts b/packages/ui-server/tests/account-writes-offloop.test.ts index 9d311e213c..ab6b4b917c 100644 --- a/packages/ui-server/tests/account-writes-offloop.test.ts +++ b/packages/ui-server/tests/account-writes-offloop.test.ts @@ -127,7 +127,7 @@ test("push HTTP routes await subscription commits and key creation", async () => function passkeys(f: ReturnType) { const failures: string[] = []; - const ctx: PasskeyContext = { log: { emit: record => { failures.push(String(record.attributes?.error ?? record.body)); } }, db: f.db, auth, revoker, rateLimiter: createLoginRateLimiter(), challenges: createPasskeyChallenges(), + const ctx: PasskeyContext = { log: { enabled: () => true, emit: record => { failures.push(String(record.attributes?.error ?? record.body)); } }, db: f.db, auth, revoker, rateLimiter: createLoginRateLimiter(), challenges: createPasskeyChallenges(), webauthn: config.webauthn, allowedOrigins: ["http://localhost"] }; const deps: PasskeyDeps = { verifyRegistrationResponse: (async () => ({ verified: true, registrationInfo: { @@ -286,6 +286,7 @@ test("revocation disconnects immediately and HTTP awaits the host's push unbindi let answered = false; const response = Promise.resolve(f.app.request(`/api/auth/principals/${f.agent.id}`, json("DELETE"))).then(r => { answered = true; return r; }); await submission; + await new Promise(resolve => setImmediate(resolve)); expect(closed, "ACCOUNT_REVOCATION_DISCONNECTS_BEFORE_UNBIND_COMMIT").toEqual([f.agent.id]); expect(answered, "ACCOUNT_REVOCATION_HTTP_AWAITS_PUSH_UNBIND_COMMIT").toBe(false); release();