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 fac43d2819..6bd0bb0ff4 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:437`). + `probeBrainCliVersion(config.brainPath`, `packages/ui-server/src/app.ts:438`). `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:375-380`). + `packages/ui-server/src/app.ts:376-381`). - **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 4b56fde13e..6db5e0dada 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:877`), the API guard follows -public ceremonies (`authGuard`, `packages/ui-server/src/app.ts:898`), and the -WebSocket admission is separate (`"/ws"`, `packages/ui-server/src/app.ts:1088`). +(`createHealthRoutes`, `packages/ui-server/src/app.ts:879`), the API guard follows +public ceremonies (`authGuard`, `packages/ui-server/src/app.ts:900`), and the +WebSocket admission is separate (`"/ws"`, `packages/ui-server/src/app.ts:1090`). Static serving is conditional (`if (options.staticRoot)`, -`packages/ui-server/src/app.ts:1108-1129`). That order establishes access; +`packages/ui-server/src/app.ts:1110-1131`). 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 10e0ac9f4c..6872707f34 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:499-515`). + (`// Recovery/heartbeat only.`, `packages/ui-server/src/app.ts:501-517`). 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 5f14651de0..2f32ba96e3 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:883` | mounted **before** the guard at `authGuard(authMode`, `app.ts:898`; logout guards itself (`resolveCookiePrincipal(c, auth, deps.db)`, `auth.ts:799`) | -| Passkey management | `passkeyManagementRoutes`, `app.ts:908` | mounted **after** the guard ("Mount AFTER the auth guard") | -| WS guard | `"/ws"`, `app.ts:1088-1105` | origin → `isWsAuthorized` (returns a boolean, `auth.ts:221-237`) → capacity → upgrade | +| Auth routes | `authRoutes`, `app.ts:885` | mounted **before** the guard at `authGuard(authMode`, `app.ts:900`; logout guards itself (`resolveCookiePrincipal(c, auth, deps.db)`, `auth.ts:798`) | +| Passkey management | `passkeyManagementRoutes`, `app.ts:910` | mounted **after** the guard ("Mount AFTER the auth guard") | +| WS guard | `"/ws"`, `app.ts:1090-1107` | 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 0862522fb3..923a3b8d47 100644 --- a/docs/decisions/sqlite-write-worker.md +++ b/docs/decisions/sqlite-write-worker.md @@ -182,8 +182,26 @@ 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. 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 + 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 @@ -341,15 +359,6 @@ are in [the governed-workers decision](governed-writable-workers.md#admitted-obj the same ordering relative to later work on that worktree. Callers without a registered writer run the same procedures inline. Boot-only source recovery retains its inline catalog writes. -- **Other app-connection writers stay on the loop for now**: push - subscriptions and VAPID keys, principal revocation and administration - and passkeys. 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 - `BEGIN IMMEDIATE` locks every attached file. Moving each of them to the - worker queue ends that wait too - ([#2850](https://github.com/schlessera/brain-kit/issues/2850)). ## Rejected 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 3bd82a2916..d186c8bb06 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:898`). + (`app.use("/api/*", authGuard(`, `packages/ui-server/src/app.ts:900`). 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 fc25c598c3..5b44b723bc 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:898`) 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:900`) 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 7faabf78d1..4e67711094 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:145-283`). -- API authorization: (`app.use("/api/*", authGuard(authMode, auth, db));`, `packages/ui-server/src/app.ts:898`). +- API authorization: (`app.use("/api/*", authGuard(authMode, auth, db));`, `packages/ui-server/src/app.ts:900`). - 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 650d9bb626..2762b349cc 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:296-332`), +- D: effective path/journal resolution (`createApp`, `packages/ui-server/src/app.ts:297-333`), 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 3e6a878f27..e38a24c786 100644 --- a/packages/ui-server/src/activity/push-sender.ts +++ b/packages/ui-server/src/activity/push-sender.ts @@ -1,7 +1,7 @@ /** * The accelerator tier: web push delivery for pending notification intents. * - * VAPID keys are generated at first boot into their own table (see migration + * VAPID keys are generated at first use into their own table (see migration * 008 for why not the settings KV) and never leave the process. Payloads are * MINIMIZED — title + outcome word + a deep link — because push renders on * lock screens outside the app's auth; detail lives behind the authenticated @@ -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 = await 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 2405d31577..5cdd12e291 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). */ @@ -102,7 +102,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 25df7e192a..8dfacb4b82 100644 --- a/packages/ui-server/src/app.ts +++ b/packages/ui-server/src/app.ts @@ -93,6 +93,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"; @@ -458,6 +459,7 @@ export async function createApp(options: CreateAppOptions = {}): Promise writer.close()); @@ -898,7 +900,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..4e86e943db 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"; @@ -434,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 @@ -445,12 +444,12 @@ 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()); - applyPrincipalRevocation(revoker, revokedIds); +): Promise { + const revokedIds = await accountWrites(db).revokeAll(Date.now()); + await applyPrincipalRevocation(revoker, revokedIds); } // --- proxy mode helpers --- @@ -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( + await 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..fd29bfe355 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()) - ); + 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 8450b4a36f..09e1669ed7 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,15 +190,15 @@ 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); } - applyPrincipalRevocation(ctx.revoker, revokedIds); + await applyPrincipalRevocation(ctx.revoker, revokedIds); ctx.referenceAuthorityRevoked?.({ grantIds: [], principalIds: revokedIds }); ctx.log?.emit({ severityText: "INFO", @@ -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 b933fafde1..4caefc49a9 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; let flushing: Promise | null = null, sweeping: Promise | null = null; diff --git a/packages/ui-server/src/ws/host.ts b/packages/ui-server/src/ws/host.ts index b2ddb88374..174125f40d 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. */ @@ -746,19 +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); - this.activity?.pushSender?.unbindPrincipal(principalId); + const unbound = this.activity?.pushSender?.unbindPrincipal(principalId); + 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(() => {}); } /** @@ -768,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 new file mode 100644 index 0000000000..ab6b4b917c --- /dev/null +++ b/packages/ui-server/tests/account-writes-offloop.test.ts @@ -0,0 +1,305 @@ +/** 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"; +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(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 }); + 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: revokerOverride, 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 failures: string[] = []; + 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: { + 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, failures }; +} + +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, 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", + }); + 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({ 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(); + } 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(); } +}); + +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; + 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(); + 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(); } +});