fix(ui-server): commit the Activity notifier and push bookkeeping on the write worker

.changeset/notifier-off-event-loop.md

@@ -0,0 +1,5 @@
+---
+"@schlessera/brain-ui-server": patch
+---
+
+Changed: the hosted server commits the Activity notifier's tick, intent delivery status, acknowledgements and retention, and push delivery's subscription bookkeeping on its SQLite write worker at `synchronous=FULL` instead of on the event loop (#2850). A delivery pass settles only once its writes have committed, so the next pass never resends an intent.

docs/decisions/action-notifications.md

@@ -178,10 +178,10 @@ about any deployment's zone or timezone-data version.
 
 The existing notifier drops later same-tag activity intents instead of
 incrementing an Action count
-(`function createIntent(input: {`, `packages/ui-server/src/activity/notify.ts:91-127`).
+(`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:197-276`).
+(`async deliverPending(notifier) {`, `packages/ui-server/src/activity/push-sender.ts:207-281`).
 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.

docs/decisions/async-collaboration.md

@@ -30,7 +30,7 @@ decision is meaningless: Queue items acquire leases; Actions stay pending or
 snoozed until a validated disposition. Every mutation emits its change row in
 the same transaction. Reuse the activity cursor algorithm and transactional
 snapshot discipline, rather than importing its activity-specific stream
-(`createActivityStream`, `packages/ui-server/src/activity/stream.ts:158-174`).
+(`createActivityStream`, `packages/ui-server/src/activity/stream.ts:160-176`).
 
 ## Authority and containment
 

docs/decisions/cost-tracking.md

@@ -166,7 +166,7 @@ as unpriced. The calls stay counted in the `brain.labeller.calls` metric, and a
 label run's cost is not added to the session's own total.
 
 A label run raises no failure or stuck notification (`const isPillLabel`,
-`packages/ui-server/src/activity/notify.ts:145`). A failed label only leaves a
+`packages/ui-server/src/activity/notify.ts:148`). A failed label only leaves a
 pill on its fallback, and its failure intent would carry the session's tag,
 `failure:<sessionId>`, so the session's next real turn failure would be
 deduplicated against it and never delivered.

docs/decisions/scheduled-tasks.md

@@ -40,8 +40,8 @@ Source inspected at `b5c30c180df5584584f261b446c4d440380d2441`:
   are available precedents; their existing tests cannot prove new schedule relations.
 - Notification kinds are already failure/completion/stuck
   (`IntentKind`, `packages/ui-server/src/activity/notify.ts:30`), with bounded
-  suppression/retry (`MAX_INTENTS_PER_HOUR`, `packages/ui-server/src/activity/notify.ts:68-74`).
-  Job-name success opt-in (`if (span.outcome === "success"`, `packages/ui-server/src/activity/notify.ts:171-184`)
+  suppression/retry (`MAX_INTENTS_PER_HOUR`, `packages/ui-server/src/activity/notify.ts:71-77`).
+  Job-name success opt-in (`if (span.outcome === "success"`, `packages/ui-server/src/activity/notify.ts:174-187`)
   does not implement schedule/occurrence preferences or correlation.
 
 Production execution requires [#689](https://github.com/schlessera/brain-kit/issues/689)

docs/decisions/sqlite-write-worker.md

@@ -185,10 +185,21 @@ are in [the governed-workers decision](governed-writable-workers.md#admitted-obj
   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.
+- **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
+  bookkeeping commit on the worker (`packages/ui-server/src/activity/notify-writes.ts`).
+  The tick runs the same notifier on the write connection, with who is
+  watching taken where the tick was submitted, and its log records come back
+  to the app's log; a tick still committing is not submitted again. A push
+  delivery pass settles only once its bookkeeping and delivery status have
+  committed, so the next pass never sends an intent again or retries a gone
+  device. The inbox and the pending queue are still read on the app
+  connection.
 - **Other app-connection writers stay on the loop for now**: retained
   retries and local exchanges, saved worker profiles, policy
-  proposals, the Activity notifier and push ticks, principal revocation and
-  administration, passkeys, inbox and settings. Each commits at
+  proposals, action and storage notices, push subscriptions and VAPID keys,
+  principal revocation and administration, passkeys, inbox and settings. Each commits at
   FULL on the loop, and while a worker procedure holds its turn, such a write
   also waits there for the worker's commit to finish, including a journal
   procedure that writes only `journal.db`, because the ledger's

docs/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:321-370`).
+(`deliverActions`, `packages/ui-server/src/activity/push-sender.ts:326-377`).
 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

docs/plans/async-collaboration-requirements.md

@@ -222,7 +222,7 @@ Stated before the requirements because five of them derive from it.
 ### Execution
 
 - R18. **The drain ticks in-server.** (`// manual triggers (triggerJob)`, `packages/ui-server/src/cron/scheduler.ts:11-14`) owns only manual triggers and
-  history, but (`export function createActivityRuntime(`, `packages/ui-server/src/activity/runtime.ts:64-203`) already runs an interval tick with a boot sweep —
+  history, but (`export function createActivityRuntime(`, `packages/ui-server/src/activity/runtime.ts:65-207`) already runs an interval tick with a boot sweep —
   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
@@ -386,8 +386,8 @@ Stated before the requirements because five of them derive from it.
 - R50. **Escalation notices need an aggregate row.** `notification_intents` constrains `kind`
   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:91-127`), and the sender emits one push per
-  pending row (`async deliverPending(notifier) {`, `packages/ui-server/src/activity/push-sender.ts:197-244`). Reuse subscriptions, retry budget, and delivery
+  (`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:207-254`). 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)

docs/plans/async-collaboration.md

@@ -127,7 +127,7 @@ system gate. It is not a prerequisite to rebuild or a new independent package.
 - **Cursors and transactional snapshots:** (`CREATE TABLE IF NOT EXISTS activity_changes (`, `packages/ui-server/migrations/007_activity.sql:58-66`)
   and (`snapshotRun(runId) {`, `packages/ui-server/src/activity/store.ts:894-923`). Reuse the algorithm; the
   activity stream is bound to activity store methods and frame types.
-- **Runtime lifecycle:** (`export function createActivityRuntime(`, `packages/ui-server/src/activity/runtime.ts:64-203`). Cron's scheduler
+- **Runtime lifecycle:** (`export function createActivityRuntime(`, `packages/ui-server/src/activity/runtime.ts:65-207`). Cron's scheduler
   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.
@@ -150,8 +150,8 @@ system gate. It is not a prerequisite to rebuild or a new independent package.
   (`export function sumEffectiveCost(`, `packages/ui-server/src/activity/store.ts:179-194`) explicitly counts unpriced runs.
   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:91-127`) and
-  (`async deliverPending(notifier) {`, `packages/ui-server/src/activity/push-sender.ts:197-244`) are run-bound and do not
+  (`function createIntent(input: {`, `packages/ui-server/src/activity/notify.ts:94-130`) and
+  (`async deliverPending(notifier) {`, `packages/ui-server/src/activity/push-sender.ts:207-254`) 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:121`) assigns the source in server code.
   (`export const SHARE_MAX_TEXT_BYTES =`, `packages/ui-sdk/src/protocol.ts:2348`) bounds text, not binary uploads;
@@ -632,7 +632,7 @@ write-once outcome discipline, cursor emission); migration comment style of `007
 - Test: `packages/ui-server/tests/inbox-runtime.test.ts`
 
 **Approach:**
-- 60s interval with a `close()` lifecycle, modeled on (`export function createActivityRuntime(`, `packages/ui-server/src/activity/runtime.ts:64-203`), including
+- 60s interval with a `close()` lifecycle, modeled on (`export function createActivityRuntime(`, `packages/ui-server/src/activity/runtime.ts:65-207`), including
   its boot sweep
 - **Gate first**: a SQL count of ready items; zero means return immediately — no model call and
   **no Activity run** (AE2)
@@ -937,7 +937,7 @@ produces exactly one executable item
 - Reuse the **algorithm** from `activity/stream.ts`, not the module — it is bound to
   activity store methods and frames (R2)
 
-**Patterns to follow:** (`export function createActivityStream(`, `packages/ui-server/src/activity/stream.ts:158-318`) poll/pump structure and chunking caps
+**Patterns to follow:** (`export function createActivityStream(`, `packages/ui-server/src/activity/stream.ts:160-320`) poll/pump structure and chunking caps
 
 **Test scenarios:**
 - Happy path: snapshot then deltas equals a fresh snapshot

packages/ui-server/src/activity/notify-writes.ts

@@ -0,0 +1,91 @@
+/**
+ * The Activity notifier and push delivery bookkeeping off the event loop
+ * (#2850). The hosted server's notifier tick (intent detection under its
+ * cursor, the stuck-run watchdog) and the writes push delivery makes
+ * (delivery status, a subscription's `last_used_at`, a gone subscription)
+ * commit on the `ui.db` write worker at `synchronous=FULL`. The tick runs the
+ * same notifier against the write connection, with who is watching taken
+ * where the tick was submitted; its log records come back to the app's log.
+ * Reads (the inbox, the pending queue) stay on the app connection.
+ */
+import type { Database } from "bun:sqlite";
+import type { LogRecord, Logger } from "@opentelemetry/api-logs";
+
+import { procedureCaller, type SqliteWriter, type WriteProcedures } from "../db/write-worker.js";
+import { createActivityNotifier, type ActivityNotifier } from "./notify.js";
+import { createActivityStore } from "./store.js";
+
+export const url = import.meta.url;
+
+/** Who was watching when a tick was submitted: `ActivityStream.isWatched`, as data. */
+export interface WatchedScopes { index: boolean; runs: string[]; sessions: string[] }
+
+export function watchedBy(scopes: WatchedScopes) {
+  return (scope: { runId?: string; sessionId?: string }) => scopes.index
+    || (scope.runId !== undefined && scopes.runs.includes(scope.runId))
+    || (scope.sessionId !== undefined && scopes.sessions.includes(scope.sessionId));
+}
+
+export function setup(db: Database): WriteProcedures {
+  let watched: WatchedScopes = { index: false, runs: [], sessions: [] };
+  const records: LogRecord[] = [];
+  // Reads only: the notifier joins terminal changes and stuck runs on this connection.
+  const notifier = createActivityNotifier({ db, store: createActivityStore(db), isWatched: scope => watchedBy(watched)(scope),
+    log: { emit: (record: LogRecord) => { records.push(record); } } as unknown as Logger });
+  return {
+    tick(now: number, scopes: WatchedScopes) {
+      watched = scopes;
+      try { notifier.tick(now); } finally { watched = { index: false, runs: [], sessions: [] }; }
+      return records.splice(0);
+    },
+    markDelivered: (id: number, status: "sent" | "send_failed") => notifier.markDelivered(id, status),
+    acknowledge: (id: number) => notifier.acknowledge(id),
+    acknowledgeAll: () => notifier.acknowledgeAll(),
+    pruneAcknowledged: (olderThanMs: number) => notifier.pruneAcknowledged(olderThanMs),
+    touchSubscription(endpoint: string, at: number, principalId?: string) {
+      if (principalId === undefined) db.query("UPDATE push_subscriptions SET last_used_at = ? WHERE endpoint = ?").run(at, endpoint);
+      else db.query("UPDATE push_subscriptions SET last_used_at = ? WHERE endpoint = ? AND principal_id = ?").run(at, endpoint, principalId);
+    },
+    dropSubscription(endpoint: string) { db.query("DELETE FROM push_subscriptions WHERE endpoint = ?").run(endpoint); },
+  };
+}
+
+/** Push delivery's subscription bookkeeping; each settles once committed. */
+export interface PushWrites {
+  touch(endpoint: string, at: number, principalId?: string): Promise<void>;
+  drop(endpoint: string): Promise<void>;
+}
+
+/**
+ * The hosted server's notifier: reads on `db`, every write on `writer`. A
+ * tick that is still committing is not submitted again; the next one runs
+ * after it.
+ */
+export function createOffloopNotifier(db: Database, writer: SqliteWriter, options: { watched: () => WatchedScopes; log?: Logger }):
+  { notifier: ActivityNotifier; push: PushWrites } {
+  const call = procedureCaller(db, { url, setup }, undefined, writer);
+  const reads = createActivityNotifier({ db, store: createActivityStore(db), isWatched: () => false, log: options.log });
+  let ticking: Promise<void> | undefined;
+  const failed = (error: unknown) => options.log?.emit({ severityText: "WARN", body: "notifier tick failed",
+    attributes: { error: error instanceof Error ? error.message : String(error) } });
+  const notifier: ActivityNotifier = {
+    tick(now) {
+      if (ticking) return ticking;
+      ticking = call<LogRecord[]>("tick", now ?? Date.now(), options.watched())
+        .then(records => { for (const record of records) options.log?.emit(record); }, failed)
+        .finally(() => { ticking = undefined; });
+      return ticking;
+    },
+    inbox: limit => reads.inbox(limit),
+    pending: limit => reads.pending(limit),
+    markDelivered: (id, status) => call<void>("markDelivered", id, status),
+    acknowledge: id => call<boolean>("acknowledge", id),
+    acknowledgeAll: () => call<number>("acknowledgeAll"),
+    pruneAcknowledged: olderThanMs => call<number>("pruneAcknowledged", olderThanMs),
+  };
+  const push: PushWrites = {
+    touch: (endpoint, at, principalId) => call<void>("touchSubscription", endpoint, at, principalId),
+    drop: endpoint => call<void>("dropSubscription", endpoint),
+  };
+  return { notifier, push };
+}

packages/ui-server/src/activity/notify.ts

@@ -51,17 +51,20 @@ export interface ActivityNotifierDeps {
 }
 
 export interface ActivityNotifier {
-  /** Detect new terminal failures + stuck runs; create intents. */
-  tick(now?: number): void;
+  /**
+   * Detect new terminal failures + stuck runs; create intents. Off the event
+   * loop (#2850), each write below settles once it has committed.
+   */
+  tick(now?: number): void | Promise<void>;
   /** Unacknowledged intents, newest first (the inbox). */
   inbox(limit?: number): NotificationIntent[];
   /** Intents awaiting delivery (the push sender's queue). */
   pending(limit?: number): NotificationIntent[];
-  markDelivered(id: number, status: "sent" | "send_failed"): void;
-  acknowledge(id: number): boolean;
-  acknowledgeAll(): number;
+  markDelivered(id: number, status: "sent" | "send_failed"): void | Promise<void>;
+  acknowledge(id: number): boolean | Promise<boolean>;
+  acknowledgeAll(): number | Promise<number>;
   /** Retention: delete acknowledged intents untouched for longer than this. */
-  pruneAcknowledged(olderThanMs: number): number;
+  pruneAcknowledged(olderThanMs: number): number | Promise<number>;
 }
 
 /** Global cap on intents CREATED per hour — failure-storm circuit. */