From 94b6e5a6e578ba07312bca0f62c29370b77cfc0f Mon Sep 17 00:00:00 2001
From: Vedran Burojevic <vedran.burojevic@gmail.com>
Date: Mon, 24 Aug 2026 08:59:46 +0200
Subject: [PATCH 1/5] Share timeline builds across clients within a 250ms
 window

The timeline response cache is keyed by maxSeq, so every appended event
makes it cold, and it deliberately refuses to store windows over 200 rows
(the streaming case). With several clients watching one streaming thread,
each self-paced refetch triggered its own synchronous 130-260ms rebuild,
and the rebuilds serialized on the event loop under every other request.

Add a per-params-key last-build slot to the timeline cache: inside a
250ms share window, requests at the same key share the one completed
build whatever its row count, and requests at a newer maxSeq for an
over-cap window get the prior window back (a rebuild floor). The floor
deliberately skips LRU-cacheable windows - those already coalesce at the
same key via the LRU, and their rebuilds stay eager - and never crosses
params keys, so a status flip (interrupt, completion) is never floored.

The route now records the latest-rows snapshot under the response's own
maxSeq rather than the route-computed one, so a floored response still
names exactly the revision its rows reflect and delta bookkeeping stays
coherent.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 apps/server/src/routes/threads/data.ts        |  12 +-
 .../src/services/threads/timeline-cache.ts    |  93 ++++++++++++--
 .../services/threads/timeline-cache.test.ts   | 114 +++++++++++++++---
 3 files changed, 188 insertions(+), 31 deletions(-)

diff --git a/apps/server/src/routes/threads/data.ts b/apps/server/src/routes/threads/data.ts
index 6aa5e42cce..7004a3b577 100644
--- a/apps/server/src/routes/threads/data.ts
+++ b/apps/server/src/routes/threads/data.ts
@@ -348,8 +348,9 @@ export function registerThreadDataRoutes(app: Hono, deps: AppDeps): void {
       summaryOnly,
       includeProviderUnhandledOperations,
     };
+    const paramsKey = buildThreadTimelineParamsKey(keyArgs);
     const full = timelineCache.getOrBuild(
-      buildThreadTimelineCacheKey({ ...keyArgs, maxSeq }),
+      { key: buildThreadTimelineCacheKey({ ...keyArgs, maxSeq }), paramsKey },
       () => {
         const { profile, response } = buildThreadTimelineWithProfile(
           deps.db,
@@ -392,7 +393,6 @@ export function registerThreadDataRoutes(app: Hono, deps: AppDeps): void {
       query.afterSequence,
       "afterSequence",
     );
-    const paramsKey = buildThreadTimelineParamsKey(keyArgs);
     const previous =
       afterSequence === undefined
         ? undefined
@@ -401,7 +401,13 @@ export function registerThreadDataRoutes(app: Hono, deps: AppDeps): void {
       previous === undefined
         ? undefined
         : computeTimelineRowDelta(previous.rows, full.rows);
-    timelineLatestRowsCache.set(paramsKey, { maxSeq, rows: full.rows });
+    // Keyed by the response's own revision, not the route-computed `maxSeq`:
+    // the cache's rebuild floor may serve a window built just before the
+    // newest events, and the snapshot must name the revision the rows reflect.
+    timelineLatestRowsCache.set(paramsKey, {
+      maxSeq: full.maxSeq,
+      rows: full.rows,
+    });
 
     return context.json(
       delta === undefined ? full : { ...full, rows: [], delta },
diff --git a/apps/server/src/services/threads/timeline-cache.ts b/apps/server/src/services/threads/timeline-cache.ts
index f5e1c903ae..e90bb740c7 100644
--- a/apps/server/src/services/threads/timeline-cache.ts
+++ b/apps/server/src/services/threads/timeline-cache.ts
@@ -22,52 +22,112 @@ import type { ThreadTimelinePageRequest } from "./timeline-pagination.js";
  * (`pruneResolvedItemDeltas`, background-task progress) is output-preserving
  * and never lowers `maxSeq`, so it cannot stale a cached entry.
  *
- * Entries with many rows are not cached: an expanded active turn (the streaming
- * case) produces hundreds of rows AND a `maxSeq` that changes on every event,
- * so caching it only thrashes the LRU and pins large objects for no reuse. Idle
- * windows collapse completed turns to a handful of rows regardless of thread
- * size, so the cap excludes exactly the entries that would never be reused.
+ * Entries with many rows are not stored in the LRU: an expanded active turn
+ * (the streaming case) produces hundreds of rows AND a `maxSeq` that changes on
+ * every event, so retaining it only thrashes the LRU and pins large objects for
+ * no reuse. Idle windows collapse completed turns to a handful of rows
+ * regardless of thread size, so the cap excludes exactly the entries that would
+ * never be reused.
+ *
+ * Those uncached streaming windows are instead shared through a short-lived
+ * last-build slot per params key (the key minus `maxSeq`). Several clients
+ * watch the same streaming thread (desktop, browser, phone, plugin panes), each
+ * on its own refetch pacing, so one appended event fans out into several
+ * back-to-back synchronous rebuilds of near-identical windows — they serialize
+ * on the event loop and stall every other request. The slot serves them all
+ * from one build:
+ *
+ * - Same key inside {@link DEFAULT_SHARE_WINDOW_MS}: the build is shared
+ *   whatever its row count. Identical key means identical output (the build is
+ *   deterministic), so this is invisible except in build count. The build is
+ *   synchronous, so it completes before the next request can start; sharing the
+ *   completed result is what coalesces loop-serialized concurrent requests.
+ * - New `maxSeq` inside the window: the prior window is returned as-is — a
+ *   rebuild floor. The response carries its own `maxSeq`, so the client simply
+ *   sees the thread as of ≤{@link DEFAULT_SHARE_WINDOW_MS} ago, below client
+ *   refetch pacing. The floor applies ONLY to responses over the LRU row cap:
+ *   LRU-cacheable windows already coalesce same-key storms via the LRU, and
+ *   keeping their rebuilds eager preserves per-request freshness for every
+ *   window the LRU can serve. `thread.status` lives in the params key, so a
+ *   status flip (interrupt, completion) is never floored.
  */
 
 const DEFAULT_MAX_ENTRIES = 128;
 const DEFAULT_MAX_CACHEABLE_ROWS = 200;
+const DEFAULT_SHARE_WINDOW_MS = 250;
 
 interface ThreadTimelineCacheOptions {
   maxEntries?: number;
-  /** Responses with more rows than this are returned but not stored. */
+  /** Responses with more rows than this are returned but not LRU-stored. */
   maxCacheableRows?: number;
+  /** How long one build is shared across requests of the same params key. */
+  shareWindowMs?: number;
+  /** Clock override for tests; defaults to Date.now. */
+  now?: () => number;
+}
+
+interface ThreadTimelineCacheKeys {
+  /** Full cache key including `maxSeq` ({@link buildThreadTimelineCacheKey}). */
+  key: string;
+  /** The key minus `maxSeq` ({@link buildThreadTimelineParamsKey}). */
+  paramsKey: string;
 }
 
 interface ThreadTimelineCache {
   getOrBuild(
-    key: string,
+    keys: ThreadTimelineCacheKeys,
     build: () => ThreadTimelineResponse,
   ): ThreadTimelineResponse;
-  /** Number of currently cached entries (for tests/metrics). */
+  /** Number of currently cached LRU entries (for tests/metrics). */
   readonly size: number;
 }
 
+interface RecentBuild {
+  key: string;
+  value: ThreadTimelineResponse;
+  builtAt: number;
+}
+
 export function createThreadTimelineCache(
   options: ThreadTimelineCacheOptions = {},
 ): ThreadTimelineCache {
   const maxEntries = options.maxEntries ?? DEFAULT_MAX_ENTRIES;
   const maxCacheableRows =
     options.maxCacheableRows ?? DEFAULT_MAX_CACHEABLE_ROWS;
+  const shareWindowMs = options.shareWindowMs ?? DEFAULT_SHARE_WINDOW_MS;
+  const now = options.now ?? Date.now;
   const entries = new Map<string, ThreadTimelineResponse>();
+  // Requests within a params key arrive with monotonic `maxSeq` (the route
+  // reads the current high-water mark), so one slot per params key holding the
+  // newest build is enough — no request can want an older revision. Expired
+  // slots are swept on every store, so the map holds at most the few windows
+  // built in the last share window.
+  const recentBuilds = new Map<string, RecentBuild>();
 
   return {
-    getOrBuild(key, build) {
-      const cached = entries.get(key);
+    getOrBuild(keys, build) {
+      const cached = entries.get(keys.key);
       if (cached !== undefined) {
         // Re-insert to mark most-recently-used.
-        entries.delete(key);
-        entries.set(key, cached);
+        entries.delete(keys.key);
+        entries.set(keys.key, cached);
         return cached;
       }
 
+      const at = now();
+      const recent = recentBuilds.get(keys.paramsKey);
+      if (recent !== undefined && at - recent.builtAt <= shareWindowMs) {
+        if (
+          recent.key === keys.key ||
+          recent.value.rows.length > maxCacheableRows
+        ) {
+          return recent.value;
+        }
+      }
+
       const value = build();
       if (value.rows.length <= maxCacheableRows) {
-        entries.set(key, value);
+        entries.set(keys.key, value);
         while (entries.size > maxEntries) {
           const oldest = entries.keys().next().value;
           if (oldest === undefined) {
@@ -76,6 +136,13 @@ export function createThreadTimelineCache(
           entries.delete(oldest);
         }
       }
+      const builtAt = now();
+      recentBuilds.set(keys.paramsKey, { key: keys.key, value, builtAt });
+      for (const [paramsKey, slot] of recentBuilds) {
+        if (builtAt - slot.builtAt > shareWindowMs) {
+          recentBuilds.delete(paramsKey);
+        }
+      }
       return value;
     },
     get size() {
diff --git a/apps/server/test/services/threads/timeline-cache.test.ts b/apps/server/test/services/threads/timeline-cache.test.ts
index 141af64fec..27bec39b8a 100644
--- a/apps/server/test/services/threads/timeline-cache.test.ts
+++ b/apps/server/test/services/threads/timeline-cache.test.ts
@@ -57,13 +57,18 @@ const baseKeyArgs: ThreadTimelineCacheKeyArgs = {
   includeProviderUnhandledOperations: false,
 };
 
+/** Give each key its own params key to exercise the LRU in isolation. */
+function keys(key: string): { key: string; paramsKey: string } {
+  return { key, paramsKey: `params:${key}` };
+}
+
 describe("createThreadTimelineCache", () => {
   it("builds once for the same key and serves cached on repeat", () => {
     const cache = createThreadTimelineCache();
     const build = vi.fn(() => makeResponse(3));
 
-    const first = cache.getOrBuild("k", build);
-    const second = cache.getOrBuild("k", build);
+    const first = cache.getOrBuild(keys("k"), build);
+    const second = cache.getOrBuild(keys("k"), build);
 
     expect(build).toHaveBeenCalledTimes(1);
     expect(second).toBe(first);
@@ -74,40 +79,119 @@ describe("createThreadTimelineCache", () => {
     const cache = createThreadTimelineCache();
     const build = vi.fn(() => makeResponse(3));
 
-    cache.getOrBuild("k1", build);
-    cache.getOrBuild("k2", build);
+    cache.getOrBuild(keys("k1"), build);
+    cache.getOrBuild(keys("k2"), build);
 
     expect(build).toHaveBeenCalledTimes(2);
   });
 
-  it("does not cache responses above the row cap (streaming expanded turns)", () => {
-    const cache = createThreadTimelineCache({ maxCacheableRows: 5 });
+  it("does not retain responses above the row cap beyond the share window", () => {
+    let nowMs = 0;
+    const cache = createThreadTimelineCache({
+      maxCacheableRows: 5,
+      shareWindowMs: 250,
+      now: () => nowMs,
+    });
     const build = vi.fn(() => makeResponse(50));
 
-    cache.getOrBuild("k", build);
-    cache.getOrBuild("k", build);
+    cache.getOrBuild(keys("k"), build);
+    nowMs += 251;
+    cache.getOrBuild(keys("k"), build);
 
     expect(build).toHaveBeenCalledTimes(2);
     expect(cache.size).toBe(0);
   });
 
   it("evicts least-recently-used entries beyond maxEntries", () => {
-    const cache = createThreadTimelineCache({ maxEntries: 2 });
+    let nowMs = 0;
+    const cache = createThreadTimelineCache({ maxEntries: 2, now: () => nowMs });
     const build = vi.fn(() => makeResponse(1));
 
-    cache.getOrBuild("a", build); // [a]
-    cache.getOrBuild("b", build); // [a,b]
-    cache.getOrBuild("a", build); // touch a -> [b,a]
-    cache.getOrBuild("c", build); // evict b -> [a,c]
+    cache.getOrBuild(keys("a"), build); // [a]
+    cache.getOrBuild(keys("b"), build); // [a,b]
+    cache.getOrBuild(keys("a"), build); // touch a -> [b,a]
+    cache.getOrBuild(keys("c"), build); // evict b -> [a,c]
 
     expect(cache.size).toBe(2);
+    // Step past the share window so the LRU alone answers the reprobe.
+    nowMs += 300;
     const buildAgain = vi.fn(() => makeResponse(1));
-    cache.getOrBuild("a", buildAgain); // still cached
-    cache.getOrBuild("b", buildAgain); // evicted -> rebuild
+    cache.getOrBuild(keys("a"), buildAgain); // still cached
+    cache.getOrBuild(keys("b"), buildAgain); // evicted -> rebuild
     expect(buildAgain).toHaveBeenCalledTimes(1);
   });
 });
 
+describe("createThreadTimelineCache share window", () => {
+  function createSharedCache(now: () => number) {
+    return createThreadTimelineCache({
+      maxCacheableRows: 5,
+      shareWindowMs: 250,
+      now,
+    });
+  }
+
+  it("shares one build across same-key requests inside the window regardless of row count", () => {
+    let nowMs = 0;
+    const cache = createSharedCache(() => nowMs);
+    const build = vi.fn(() => makeResponse(50));
+
+    const first = cache.getOrBuild(keys("k"), build);
+    nowMs += 100;
+    const second = cache.getOrBuild(keys("k"), build);
+
+    expect(build).toHaveBeenCalledTimes(1);
+    // Same revision must mean identical rows for every client.
+    expect(second).toBe(first);
+    // Shared, not retained: the LRU row cap still applies.
+    expect(cache.size).toBe(0);
+  });
+
+  it("floors rebuilds of over-cap windows: a new maxSeq inside the window gets the prior window", () => {
+    let nowMs = 0;
+    const cache = createSharedCache(() => nowMs);
+    const build = vi.fn(() => makeResponse(50));
+
+    const first = cache.getOrBuild({ key: "10|p", paramsKey: "p" }, build);
+    nowMs += 100;
+    const floored = cache.getOrBuild({ key: "11|p", paramsKey: "p" }, build);
+    expect(build).toHaveBeenCalledTimes(1);
+    expect(floored).toBe(first);
+
+    nowMs += 200; // 300ms since the build: past the window.
+    const rebuilt = cache.getOrBuild({ key: "12|p", paramsKey: "p" }, build);
+    expect(build).toHaveBeenCalledTimes(2);
+    expect(rebuilt).not.toBe(first);
+  });
+
+  it("keeps rebuilds eager for LRU-cacheable windows (no floor below the row cap)", () => {
+    let nowMs = 0;
+    const cache = createSharedCache(() => nowMs);
+    const build = vi.fn(() => makeResponse(3));
+
+    cache.getOrBuild({ key: "10|p", paramsKey: "p" }, build);
+    nowMs += 10;
+    cache.getOrBuild({ key: "11|p", paramsKey: "p" }, build);
+
+    expect(build).toHaveBeenCalledTimes(2);
+  });
+
+  it("never floors across params keys (a status flip builds fresh)", () => {
+    let nowMs = 0;
+    const cache = createSharedCache(() => nowMs);
+    const build = vi.fn(() => makeResponse(50));
+
+    cache.getOrBuild({ key: "10|p:active", paramsKey: "p:active" }, build);
+    nowMs += 10;
+    cache.getOrBuild(
+      { key: "11|p:interrupted", paramsKey: "p:interrupted" },
+      build,
+    );
+
+    expect(build).toHaveBeenCalledTimes(2);
+  });
+});
+
 describe("buildThreadTimelineCacheKey", () => {
   it("differs when any projection input differs", () => {
     const base = buildThreadTimelineCacheKey(baseKeyArgs);

From 108eb1dd8646d1a8c2f0a60892faa977e9c0b78a Mon Sep 17 00:00:00 2001
From: Vedran Burojevic <vedran.burojevic@gmail.com>
Date: Mon, 24 Aug 2026 13:46:53 +0200
Subject: [PATCH 2/5] Defer the active-thread event prune off the ingest
 response path

maybePruneActiveThreadEventHistory ran synchronously inside
POST /internal/session/events - the hottest handler - every ~30s per
active thread, and it was not wrapped in event-loop-work, so the stall
monitor could not name it when it exceeded a second. Two of its DELETEs
also walk their candidate rows with correlated subqueries even when the
thread has nothing prunable, which is the common case.

Move the prune into the existing deferEventFollowUpBatch as an
"active-thread-event-prune" follow-up, wrapped in runEventLoopWorkSync
so stalls are attributable to `event-prune <threadId>`. Pruning is
output-preserving and never lowers maxSeq, so deferral only changes when
redundant rows disappear, never what any read returns - the updated
ingest test pins both the post-response ordering and that a timeline
read between append and deferred prune projects identical rows.

Cheapen the no-op case with a LIMIT 1 necessary-condition probe per
walking prune class (usage CTEs, resolved item deltas, background-task
progress): one covering (thread_id, type, sequence) index seek decides
whether the DELETE can possibly match before it walks anything. The
plain range-delete class needs no probe - it already is one.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 apps/server/src/internal/events.ts            |  28 ++++-
 .../src/services/system/event-pruning.ts      |  97 +++++++++++++++--
 apps/server/test/system/event-pruning.test.ts | 100 +++++++++++++++++-
 3 files changed, 214 insertions(+), 11 deletions(-)

diff --git a/apps/server/src/internal/events.ts b/apps/server/src/internal/events.ts
index 9fde9228b7..394794c83e 100644
--- a/apps/server/src/internal/events.ts
+++ b/apps/server/src/internal/events.ts
@@ -37,6 +37,7 @@ import {
   isActivePruneTriggerThreadEventType,
   maybePruneActiveThreadEventHistory,
 } from "../services/system/event-pruning.js";
+import { runEventLoopWorkSync } from "../services/system/event-loop-work.js";
 import { queueChildThreadTurnNotificationBestEffort } from "../services/threads/child-thread-notifications.js";
 import { isParentNotifiableChildThread } from "../services/threads/thread-parent.js";
 import { runQueuedMessageAutoSendForThread } from "../services/threads/queued-messages.js";
@@ -194,9 +195,22 @@ interface QueuedMessageAutoSendFollowUp {
   threadId: string;
 }
 
+/**
+ * Opportunistic event-history prune, deferred off the ingest response path.
+ * Pruning is output-preserving (see the timeline cache notes), so running it
+ * after the response — instead of inside the hottest handler — only changes
+ * when the redundant rows disappear, never what any read returns.
+ */
+interface ActiveThreadEventPruneFollowUp {
+  kind: "active-thread-event-prune";
+  latestPrunableSequence: number;
+  threadId: string;
+}
+
 type EventEffectFollowUp =
   | ParentTurnNotificationFollowUp
-  | QueuedMessageAutoSendFollowUp;
+  | QueuedMessageAutoSendFollowUp
+  | ActiveThreadEventPruneFollowUp;
 
 function isRootTurnStartedEvent(
   event: Extract<HostDaemonEventEnvelope["event"], { type: "turn/started" }>,
@@ -494,6 +508,16 @@ async function executeEventFollowUpBestEffort(
           threadId: followUp.threadId,
         });
         return;
+      case "active-thread-event-prune":
+        // Synchronous better-sqlite3 work: the wrapper lets the stall
+        // monitor name the prune instead of blaming whatever ran next.
+        runEventLoopWorkSync(`event-prune ${followUp.threadId}`, () =>
+          maybePruneActiveThreadEventHistory(deps, {
+            latestPrunableSequence: followUp.latestPrunableSequence,
+            threadId: followUp.threadId,
+          }),
+        );
+        return;
     }
   } catch (error) {
     if (isCommandTimeoutError(error)) {
@@ -1023,7 +1047,7 @@ export function registerInternalEventRoutes(app: Hono, deps: AppDeps): void {
         events: postableEvents,
         insertedEventIndexes: appendResult.insertedInputIndexes,
       })) {
-        maybePruneActiveThreadEventHistory(deps, candidate);
+        followUps.push({ kind: "active-thread-event-prune", ...candidate });
       }
 
       deferEventFollowUpBatch(deps, followUps);
diff --git a/apps/server/src/services/system/event-pruning.ts b/apps/server/src/services/system/event-pruning.ts
index 2f584750e5..277b82643d 100644
--- a/apps/server/src/services/system/event-pruning.ts
+++ b/apps/server/src/services/system/event-pruning.ts
@@ -1,4 +1,5 @@
 import { performance } from "node:perf_hooks";
+import { and, eq, inArray, lte } from "drizzle-orm";
 import {
   getThread,
   getLatestThreadSequence,
@@ -7,6 +8,7 @@ import {
   pruneResolvedItemDeltas,
   pruneTokenUsageEventsBeforeSequence,
   pruneThreadEventsBeforeSequence,
+  events as storedEvents,
 } from "@bb/db";
 import type { ThreadEventType } from "@bb/domain";
 import { roundDurationMs } from "../lib/duration.js";
@@ -85,6 +87,18 @@ const GENERIC_AGE_PRUNABLE_THREAD_EVENT_TYPES: readonly ThreadEventType[] = [
   "turn/diff/updated",
 ] as const;
 
+/**
+ * Candidate types for `pruneResolvedItemDeltas`. Must stay a superset of the
+ * delta types that DELETE targets, or the probe below silently disables the
+ * prune for the missing type.
+ */
+const RESOLVED_ITEM_DELTA_THREAD_EVENT_TYPES: readonly ThreadEventType[] = [
+  "item/agentMessage/delta",
+  "item/commandExecution/outputDelta",
+  "item/reasoning/summaryTextDelta",
+  "item/reasoning/textDelta",
+] as const;
+
 const KEEP_RECENT_BY_MODE: Record<ThreadEventPruningMode, number> = {
   active: ACTIVE_THREAD_EVENT_KEEP_RECENT,
   idle: IDLE_THREAD_EVENT_KEEP_RECENT,
@@ -125,6 +139,49 @@ export function isActivePruneTriggerThreadEventType(
   return activePruneTriggerThreadEventTypeSet.has(eventType);
 }
 
+interface HasPrunableCandidateRowsArgs {
+  threadId: string;
+  types: readonly ThreadEventType[];
+  /** Omit for classes whose DELETE has no sequence bound. */
+  sequenceCutoff?: number;
+}
+
+/**
+ * LIMIT 1 necessary-condition probe for a prune class. The usage-row DELETEs
+ * walk every row of their type through a correlated-subquery CTE, and the
+ * resolved-delta/background-progress DELETEs walk their candidates with
+ * correlated EXISTS checks — even when the thread has no candidate rows at
+ * all, which is the common case for an active thread pruned every ~30s. This
+ * probe answers "could that DELETE possibly match?" with one covering seek on
+ * the (thread_id, type, sequence) index, so the no-op case skips the walk
+ * outright. A true result only means candidates exist; the DELETE still
+ * decides what is actually prunable.
+ */
+function hasPrunableCandidateRows(
+  db: AppDeps["db"],
+  args: HasPrunableCandidateRowsArgs,
+): boolean {
+  if (args.sequenceCutoff !== undefined && args.sequenceCutoff <= 0) {
+    return false;
+  }
+  return (
+    db
+      .select({ id: storedEvents.id })
+      .from(storedEvents)
+      .where(
+        and(
+          eq(storedEvents.threadId, args.threadId),
+          inArray(storedEvents.type, [...args.types]),
+          ...(args.sequenceCutoff === undefined
+            ? []
+            : [lte(storedEvents.sequence, args.sequenceCutoff)]),
+        ),
+      )
+      .limit(1)
+      .get() !== undefined
+  );
+}
+
 export function pruneThreadEventHistory(
   deps: Pick<AppDeps, "db">,
   args: PruneThreadEventHistoryArgs,
@@ -140,17 +197,31 @@ export function pruneThreadEventHistory(
   const sequenceCutoff = Math.max(0, latestSequence - keepRecent);
   const removedAgePrunableEvents =
     runThreadEventPruningStep("prune_context_window_usage", () =>
-      pruneContextWindowUsageEventsBeforeSequence(deps.db, {
+      hasPrunableCandidateRows(deps.db, {
         threadId: args.threadId,
+        types: ["thread/contextWindowUsage/updated"],
         sequenceCutoff,
-      }),
+      })
+        ? pruneContextWindowUsageEventsBeforeSequence(deps.db, {
+            threadId: args.threadId,
+            sequenceCutoff,
+          })
+        : 0,
     ) +
     runThreadEventPruningStep("prune_token_usage", () =>
-      pruneTokenUsageEventsBeforeSequence(deps.db, {
+      hasPrunableCandidateRows(deps.db, {
         threadId: args.threadId,
+        types: ["thread/tokenUsage/updated"],
         sequenceCutoff,
-      }),
+      })
+        ? pruneTokenUsageEventsBeforeSequence(deps.db, {
+            threadId: args.threadId,
+            sequenceCutoff,
+          })
+        : 0,
     ) +
+    // No probe: this DELETE is already a plain (thread_id, type, sequence)
+    // index-range scan, exactly what the probe would run.
     runThreadEventPruningStep("prune_generic_age_prunable_events", () =>
       pruneThreadEventsBeforeSequence(deps.db, {
         threadId: args.threadId,
@@ -161,16 +232,26 @@ export function pruneThreadEventHistory(
   const removedResolvedItemDeltas = runThreadEventPruningStep(
     "prune_resolved_item_deltas",
     () =>
-      pruneResolvedItemDeltas(deps.db, {
+      hasPrunableCandidateRows(deps.db, {
         threadId: args.threadId,
-      }),
+        types: RESOLVED_ITEM_DELTA_THREAD_EVENT_TYPES,
+      })
+        ? pruneResolvedItemDeltas(deps.db, {
+            threadId: args.threadId,
+          })
+        : 0,
   );
   const removedBackgroundTaskProgressEvents = runThreadEventPruningStep(
     "prune_background_task_progress",
     () =>
-      pruneBackgroundTaskProgressEvents(deps.db, {
+      hasPrunableCandidateRows(deps.db, {
         threadId: args.threadId,
-      }),
+        types: ["item/backgroundTask/progress"],
+      })
+        ? pruneBackgroundTaskProgressEvents(deps.db, {
+            threadId: args.threadId,
+          })
+        : 0,
   );
 
   return {
diff --git a/apps/server/test/system/event-pruning.test.ts b/apps/server/test/system/event-pruning.test.ts
index 5ebe261177..06b1854618 100644
--- a/apps/server/test/system/event-pruning.test.ts
+++ b/apps/server/test/system/event-pruning.test.ts
@@ -1,5 +1,5 @@
 import { getThread, listEvents } from "@bb/db";
-import { turnScope } from "@bb/domain";
+import { threadScope, turnScope } from "@bb/domain";
 import { groupHostDaemonEvents } from "@bb/host-daemon-contract";
 import { describe, expect, it, vi } from "vitest";
 import { applyTurnCompletedEvent } from "../../src/internal/turn-completed-events.js";
@@ -130,6 +130,7 @@ function seedResolvedAssistantMessage(
   for (const sequence of args.deltaSequences) {
     seedStoredEvent(harness.deps, {
       threadId: args.threadId,
+      providerThreadId: "provider-thread-1",
       sequence,
       type: "item/agentMessage/delta",
       scope: turnScope(turnId),
@@ -144,6 +145,7 @@ function seedResolvedAssistantMessage(
 
   seedStoredEvent(harness.deps, {
     threadId: args.threadId,
+    providerThreadId: "provider-thread-1",
     sequence: args.completedSequence,
     type: "item/completed",
     scope: turnScope(turnId),
@@ -532,6 +534,7 @@ describe("thread event pruning", () => {
       for (const sequence of [1_004, 1_005]) {
         seedStoredEvent(harness.deps, {
           threadId: thread.id,
+          providerThreadId: "provider-thread-1",
           scope: turnScope("turn-active"),
           sequence,
           type: "item/agentMessage/delta",
@@ -580,6 +583,46 @@ describe("thread event pruning", () => {
       });
 
       expect(response.status).toBe(200);
+
+      // The prune is deferred off the response path: when the response
+      // settles, nothing has been removed yet.
+      expect(
+        listEventSequencesForType(harness, {
+          threadId: thread.id,
+          type: "thread/tokenUsage/updated",
+        }).at(0),
+      ).toBe(1);
+      expect(
+        listEventSequencesForType(harness, {
+          threadId: thread.id,
+          type: "item/agentMessage/delta",
+          itemId: "msg-completed",
+        }),
+      ).toEqual([1_001, 1_002]);
+
+      // Pruning is output-preserving, so a timeline read in the window
+      // between the append and the deferred prune must project exactly what
+      // a post-prune read projects.
+      const timelineOptions = {
+        eventBudget: 1_000_000,
+        includeProviderUnhandledOperations: true,
+        maxInlineOutputChars: null,
+        maxSeq: 0,
+        page: {
+          kind: "latest",
+          segmentLimit: Number.MAX_SAFE_INTEGER,
+        },
+      } as const;
+      const beforePrune = buildThreadTimeline(
+        harness.db,
+        thread,
+        timelineOptions,
+      );
+
+      // The deferred batch was scheduled with setImmediate during the
+      // request; one immediate turn later it has run.
+      await new Promise((resolve) => setImmediate(resolve));
+
       expect(
         listEventSequencesForType(harness, {
           threadId: thread.id,
@@ -600,6 +643,61 @@ describe("thread event pruning", () => {
           itemId: "msg-active",
         }),
       ).toEqual([1_004, 1_005]);
+
+      const afterPrune = buildThreadTimeline(
+        harness.db,
+        thread,
+        timelineOptions,
+      );
+      expect(afterPrune.rows).toEqual(beforePrune.rows);
+      expect(afterPrune.contextWindowUsage).toEqual(
+        beforePrune.contextWindowUsage,
+      );
+    });
+  });
+
+  it("prunes superseded background-task progress snapshots", async () => {
+    await withTestHarness(async (harness) => {
+      const host = seedHost(harness.deps);
+      const { project } = seedProjectWithSource(harness.deps, {
+        hostId: host.id,
+      });
+      const environment = seedEnvironment(harness.deps, {
+        hostId: host.id,
+        projectId: project.id,
+      });
+      const thread = seedThread(harness.deps, {
+        projectId: project.id,
+        environmentId: environment.id,
+      });
+
+      // Two snapshots for the same item: only the latest is load-bearing.
+      for (const sequence of [1, 2]) {
+        seedStoredEvent(harness.deps, {
+          threadId: thread.id,
+          sequence,
+          type: "item/backgroundTask/progress",
+          scope: threadScope(),
+          itemId: "task-1",
+          itemKind: "backgroundTask",
+          data: {
+            item: { id: "task-1", type: "backgroundTask", status: "running" },
+          },
+        });
+      }
+
+      const result = pruneThreadEventHistory(harness.deps, {
+        mode: "idle",
+        threadId: thread.id,
+      });
+
+      expect(result.removedBackgroundTaskProgressEvents).toBe(1);
+      expect(
+        listEventSequencesForType(harness, {
+          threadId: thread.id,
+          type: "item/backgroundTask/progress",
+        }),
+      ).toEqual([2]);
     });
   });
 });

From 3a9b6377ce4b326510e5f0b3bdb372c6a49d502a Mon Sep 17 00:00:00 2001
From: Vedran Burojevic <vedran.burojevic@gmail.com>
Date: Mon, 24 Aug 2026 13:54:10 +0200
Subject: [PATCH 3/5] Backpressure every app websocket fan-out through the
 terminal lane machinery

The app broadcast paths (notifyClientsByKeySet, notifyThreadListOnlySockets,
notifyThreadOpen, notifyThreadPaneAction, notifyPluginSignal) looped bare
socket.send: no bufferedAmount check, no queue cap, and no try/catch, so a
suspended phone's socket buffer grew without bound into server RSS and one
throwing socket truncated delivery to every subscriber after it.

Generalize the terminal queue/drain/drop machinery into per-lane
sendWithBackpressure: send below the socket high water, queue and
poll-drain above it, and drop-and-close the socket (1013) when a send
throws or the queue budget overflows. The terminal lane keeps its
1 MiB / 32 MiB limits and drop semantics verbatim; the new app lane uses
1 MiB / 4 MiB - app messages are small invalidation hints, and a socket
that cannot flush them is wedged. Dropping an app socket unregisters it
first so the rest of the fan-out and all later ones skip it; clients
already reconnect and catch up via the watermark. Broadcast counts now
report sockets that accepted (sent or queued) the signal.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 apps/server/src/ws/hub.ts        | 173 +++++++++++++++++++++----------
 apps/server/test/app/hub.test.ts |  77 ++++++++++++++
 2 files changed, 196 insertions(+), 54 deletions(-)

diff --git a/apps/server/src/ws/hub.ts b/apps/server/src/ws/hub.ts
index 271122f2d5..ffbf212508 100644
--- a/apps/server/src/ws/hub.ts
+++ b/apps/server/src/ws/hub.ts
@@ -35,7 +35,14 @@ const TERMINAL_SOCKET_HIGH_WATER_BYTES = 1024 * 1024;
 // A 16 MiB raw burst expands to about 21.4 MiB as base64 + JSON. Keep
 // enough bounded headroom for that workload while preventing unbounded growth.
 const TERMINAL_SOCKET_MAX_QUEUE_BYTES = 32 * 1024 * 1024;
-const TERMINAL_SOCKET_DRAIN_POLL_MS = 10;
+// App realtime messages are small invalidation hints and ephemeral signals; a
+// socket that cannot flush 1 MiB of them is wedged (a suspended phone, a dead
+// network), and letting its buffers grow just leaks server memory. Queue a
+// short burst, then drop the socket: clients reconnect and catch up via the
+// watermark.
+const APP_SOCKET_HIGH_WATER_BYTES = 1024 * 1024;
+const APP_SOCKET_MAX_QUEUE_BYTES = 4 * 1024 * 1024;
+const SOCKET_DRAIN_POLL_MS = 10;
 /**
  * A streaming turn appends events ~10 times a second. A client that only
  * subscribes to the thread list (every open app window, for every thread it
@@ -58,12 +65,27 @@ interface HubSocket {
   send(data: string): void;
 }
 
-interface TerminalSocketSendQueue {
+interface SocketSendQueue {
   bytes: number;
   payloads: string[];
   timeout: ReturnType<typeof setTimeout> | null;
 }
 
+/**
+ * One backpressured send path. The hub runs two: the terminal lane (large
+ * ordered output bursts, generous queue) and the app lane (small realtime
+ * notifications, tight queue). Both share the same mechanics — send directly
+ * below the socket high water, queue and poll-drain above it, and hand the
+ * socket to `drop` when a send throws or the queue budget overflows.
+ */
+interface SocketSendLane {
+  drop: (socket: HubSocket, reason: string) => void;
+  dropReasons: { queueOverflow: string; sendFailed: string };
+  highWaterBytes: number;
+  maxQueueBytes: number;
+  queues: Map<HubSocket, SocketSendQueue>;
+}
+
 type ChangedMessageListener = (message: ChangedMessage) => void;
 
 interface PendingThreadListEventsAppended {
@@ -225,10 +247,26 @@ export class NotificationHub implements DbNotifier {
     string,
     Set<HubSocket>
   >();
-  private readonly terminalSocketSendQueues = new Map<
-    HubSocket,
-    TerminalSocketSendQueue
-  >();
+  private readonly terminalSendLane: SocketSendLane = {
+    drop: (socket, reason) => this.dropTerminalSocket(socket, reason),
+    dropReasons: {
+      queueOverflow: "terminal-backpressure",
+      sendFailed: "terminal-send-failed",
+    },
+    highWaterBytes: TERMINAL_SOCKET_HIGH_WATER_BYTES,
+    maxQueueBytes: TERMINAL_SOCKET_MAX_QUEUE_BYTES,
+    queues: new Map<HubSocket, SocketSendQueue>(),
+  };
+  private readonly appSendLane: SocketSendLane = {
+    drop: (socket, reason) => this.dropAppSocket(socket, reason),
+    dropReasons: {
+      queueOverflow: "app-backpressure",
+      sendFailed: "app-send-failed",
+    },
+    highWaterBytes: APP_SOCKET_HIGH_WATER_BYTES,
+    maxQueueBytes: APP_SOCKET_MAX_QUEUE_BYTES,
+    queues: new Map<HubSocket, SocketSendQueue>(),
+  };
   private readonly terminalIdsByClientSocket = new Map<
     HubSocket,
     Set<string>
@@ -251,6 +289,7 @@ export class NotificationHub implements DbNotifier {
 
   unregisterClient(socket: HubSocket): void {
     this.unregisterTerminalClientSocket(socket);
+    this.clearSocketSendQueue(this.appSendLane, socket);
     const keys = this.clientKeysBySocket.get(socket);
     if (!keys) {
       return;
@@ -314,7 +353,7 @@ export class NotificationHub implements DbNotifier {
     terminalIds.delete(terminalId);
     if (terminalIds.size === 0) {
       this.terminalIdsByClientSocket.delete(socket);
-      this.clearTerminalSocketSendQueue(socket);
+      this.clearSocketSendQueue(this.terminalSendLane, socket);
     }
   }
 
@@ -337,7 +376,7 @@ export class NotificationHub implements DbNotifier {
     }
 
     this.terminalIdsByClientSocket.delete(socket);
-    this.clearTerminalSocketSendQueue(socket);
+    this.clearSocketSendQueue(this.terminalSendLane, socket);
   }
 
   private releaseTerminalResizeOwnership(
@@ -363,7 +402,8 @@ export class NotificationHub implements DbNotifier {
     socket: HubSocket,
     message: TerminalServerMessage,
   ): void {
-    this.sendOrQueueTerminalPayload(
+    this.sendWithBackpressure(
+      this.terminalSendLane,
       socket,
       JSON.stringify(terminalServerMessageSchema.parse(message)),
     );
@@ -380,22 +420,32 @@ export class NotificationHub implements DbNotifier {
 
     const payload = JSON.stringify(terminalServerMessageSchema.parse(message));
     for (const socket of [...sockets]) {
-      this.sendOrQueueTerminalPayload(socket, payload);
+      this.sendWithBackpressure(this.terminalSendLane, socket, payload);
     }
   }
 
-  private sendOrQueueTerminalPayload(socket: HubSocket, payload: string): void {
-    const existingQueue = this.terminalSocketSendQueues.get(socket);
+  /**
+   * Send on a lane, queueing above the socket high water and dropping the
+   * socket when a send throws or the queue budget overflows — one bad socket
+   * can neither stall the loop nor grow without bound. Returns false when the
+   * socket was dropped instead of sent/queued to.
+   */
+  private sendWithBackpressure(
+    lane: SocketSendLane,
+    socket: HubSocket,
+    payload: string,
+  ): boolean {
+    const existingQueue = lane.queues.get(socket);
     if (
       !existingQueue &&
-      (socket.raw?.bufferedAmount ?? 0) <= TERMINAL_SOCKET_HIGH_WATER_BYTES
+      (socket.raw?.bufferedAmount ?? 0) <= lane.highWaterBytes
     ) {
       try {
         socket.send(payload);
-        return;
+        return true;
       } catch {
-        this.dropTerminalSocket(socket, "terminal-send-failed");
-        return;
+        lane.drop(socket, lane.dropReasons.sendFailed);
+        return false;
       }
     }
 
@@ -405,39 +455,42 @@ export class NotificationHub implements DbNotifier {
       timeout: null,
     };
     const payloadBytes = Buffer.byteLength(payload, "utf8");
-    if (queue.bytes + payloadBytes > TERMINAL_SOCKET_MAX_QUEUE_BYTES) {
-      this.dropTerminalSocket(socket, "terminal-backpressure");
-      return;
+    if (queue.bytes + payloadBytes > lane.maxQueueBytes) {
+      lane.drop(socket, lane.dropReasons.queueOverflow);
+      return false;
     }
     queue.payloads.push(payload);
     queue.bytes += payloadBytes;
-    this.terminalSocketSendQueues.set(socket, queue);
-    this.scheduleTerminalSocketDrain(socket, queue);
+    lane.queues.set(socket, queue);
+    this.scheduleSocketDrain(lane, socket, queue);
+    return true;
   }
 
-  private scheduleTerminalSocketDrain(
+  private scheduleSocketDrain(
+    lane: SocketSendLane,
     socket: HubSocket,
-    queue: TerminalSocketSendQueue,
+    queue: SocketSendQueue,
   ): void {
     if (queue.timeout !== null) {
       return;
     }
     queue.timeout = setTimeout(() => {
       queue.timeout = null;
-      this.flushTerminalSocketQueue(socket, queue);
-    }, TERMINAL_SOCKET_DRAIN_POLL_MS);
+      this.flushSocketQueue(lane, socket, queue);
+    }, SOCKET_DRAIN_POLL_MS);
   }
 
-  private flushTerminalSocketQueue(
+  private flushSocketQueue(
+    lane: SocketSendLane,
     socket: HubSocket,
-    queue: TerminalSocketSendQueue,
+    queue: SocketSendQueue,
   ): void {
-    if (this.terminalSocketSendQueues.get(socket) !== queue) {
+    if (lane.queues.get(socket) !== queue) {
       return;
     }
     while (
       queue.payloads.length > 0 &&
-      (socket.raw?.bufferedAmount ?? 0) <= TERMINAL_SOCKET_HIGH_WATER_BYTES
+      (socket.raw?.bufferedAmount ?? 0) <= lane.highWaterBytes
     ) {
       const payload = queue.payloads[0];
       if (payload === undefined) {
@@ -446,17 +499,17 @@ export class NotificationHub implements DbNotifier {
       try {
         socket.send(payload);
       } catch {
-        this.dropTerminalSocket(socket, "terminal-send-failed");
+        lane.drop(socket, lane.dropReasons.sendFailed);
         return;
       }
       queue.payloads.shift();
       queue.bytes -= Buffer.byteLength(payload, "utf8");
     }
     if (queue.payloads.length === 0) {
-      this.clearTerminalSocketSendQueue(socket);
+      this.clearSocketSendQueue(lane, socket);
       return;
     }
-    this.scheduleTerminalSocketDrain(socket, queue);
+    this.scheduleSocketDrain(lane, socket, queue);
   }
 
   private dropTerminalSocket(socket: HubSocket, reason: string): void {
@@ -468,15 +521,29 @@ export class NotificationHub implements DbNotifier {
     }
   }
 
-  private clearTerminalSocketSendQueue(socket: HubSocket): void {
-    const queue = this.terminalSocketSendQueues.get(socket);
+  /**
+   * Unregister first (which also clears the socket's queues) so the rest of
+   * the current fan-out and all later ones skip the socket; the client
+   * reconnects and catches up via the watermark.
+   */
+  private dropAppSocket(socket: HubSocket, reason: string): void {
+    this.unregisterClient(socket);
+    try {
+      socket.close(1013, reason);
+    } catch {
+      // The socket is already unusable; registration and queue state are gone.
+    }
+  }
+
+  private clearSocketSendQueue(lane: SocketSendLane, socket: HubSocket): void {
+    const queue = lane.queues.get(socket);
     if (!queue) {
       return;
     }
     if (queue.timeout !== null) {
       clearTimeout(queue.timeout);
     }
-    this.terminalSocketSendQueues.delete(socket);
+    lane.queues.delete(socket);
   }
 
   subscribe(socket: HubSocket, target: RealtimeSubscriptionTarget): void {
@@ -804,12 +871,7 @@ export class NotificationHub implements DbNotifier {
         file: request.file,
       }),
     );
-    let delivered = 0;
-    for (const socket of this.clientKeysBySocket.keys()) {
-      socket.send(payload);
-      delivered += 1;
-    }
-    return delivered;
+    return this.broadcastToAppSockets(payload);
   }
 
   /** Broadcast an ephemeral pane presentation request to every app client. */
@@ -825,12 +887,7 @@ export class NotificationHub implements DbNotifier {
         action,
       }),
     );
-    let delivered = 0;
-    for (const socket of this.clientKeysBySocket.keys()) {
-      socket.send(payload);
-      delivered += 1;
-    }
-    return delivered;
+    return this.broadcastToAppSockets(payload);
   }
 
   /**
@@ -852,10 +909,17 @@ export class NotificationHub implements DbNotifier {
         payload,
       }),
     );
+    return this.broadcastToAppSockets(message);
+  }
+
+  /** Ephemeral broadcast to every app socket; returns how many accepted it. */
+  private broadcastToAppSockets(payload: string): number {
     let delivered = 0;
-    for (const socket of this.clientKeysBySocket.keys()) {
-      socket.send(message);
-      delivered += 1;
+    // Snapshot: dropping a socket mutates the live registration map.
+    for (const socket of [...this.clientKeysBySocket.keys()]) {
+      if (this.sendWithBackpressure(this.appSendLane, socket, payload)) {
+        delivered += 1;
+      }
     }
     return delivered;
   }
@@ -1055,11 +1119,11 @@ export class NotificationHub implements DbNotifier {
       return;
     }
     const detailKey = subscriptionKey({ kind: "thread-detail", threadId });
-    for (const socket of listSockets) {
+    for (const socket of [...listSockets]) {
       if (this.clientKeysBySocket.get(socket)?.has(detailKey)) {
         continue;
       }
-      socket.send(payload);
+      this.sendWithBackpressure(this.appSendLane, socket, payload);
     }
   }
 
@@ -1089,8 +1153,9 @@ export class NotificationHub implements DbNotifier {
     sockets: Iterable<HubSocket>,
     payload: string,
   ): void {
-    for (const socket of sockets) {
-      socket.send(payload);
+    // Snapshot: dropping a socket mutates the live subscription sets.
+    for (const socket of [...sockets]) {
+      this.sendWithBackpressure(this.appSendLane, socket, payload);
     }
   }
 
diff --git a/apps/server/test/app/hub.test.ts b/apps/server/test/app/hub.test.ts
index 5d8ebcee69..4dfafdfd2e 100644
--- a/apps/server/test/app/hub.test.ts
+++ b/apps/server/test/app/hub.test.ts
@@ -250,6 +250,83 @@ describe("NotificationHub", () => {
     expect(socket.messages).toEqual([]);
   });
 
+  it("queues app notifications for a socket above high water and flushes on drain", () => {
+    vi.useFakeTimers();
+    const hub = new NotificationHub();
+    const healthy = createMockHubSocket();
+    const wedged = {
+      ...createMockHubSocket(),
+      raw: { bufferedAmount: 2 * 1024 * 1024 },
+    };
+    hub.subscribe(healthy, { kind: "thread-detail", threadId: "thread-1" });
+    hub.subscribe(wedged, { kind: "thread-detail", threadId: "thread-1" });
+
+    hub.notifyThread("thread-1", ["status-changed"]);
+
+    expect(healthy.messages).toHaveLength(1);
+    expect(wedged.messages).toHaveLength(0);
+
+    wedged.raw.bufferedAmount = 0;
+    vi.advanceTimersByTime(10);
+    expect(wedged.messages).toHaveLength(1);
+  });
+
+  it("drops a wedged app socket at the queue cap while siblings still receive", () => {
+    const hub = new NotificationHub();
+    const healthy = createMockHubSocket();
+    const wedged = {
+      ...createMockHubSocket(),
+      raw: { bufferedAmount: 2 * 1024 * 1024 },
+    };
+    hub.registerClient(healthy);
+    hub.registerClient(wedged);
+    const bigPayload = "x".repeat(1024 * 1024);
+
+    for (let index = 0; index < 5; index += 1) {
+      hub.notifyPluginSignal("plugin-1", "channel-1", bigPayload);
+    }
+
+    expect(wedged.closed).toContainEqual({
+      code: 1013,
+      reason: "app-backpressure",
+    });
+    expect(wedged.messages).toEqual([]);
+    expect(healthy.messages).toHaveLength(5);
+
+    // The drop unregistered the socket: later broadcasts skip it entirely.
+    const delivered = hub.notifyPluginSignal("plugin-1", "channel-1", "after");
+    expect(delivered).toBe(1);
+    expect(wedged.closed).toHaveLength(1);
+  });
+
+  it("isolates a throwing app socket from healthy subscribers", () => {
+    const hub = new NotificationHub();
+    const healthy = createMockHubSocket();
+    const closed: Array<{ code?: number; reason?: string }> = [];
+    const failing = {
+      close(code?: number, reason?: string) {
+        closed.push({ code, reason });
+      },
+      send() {
+        throw new Error("socket send failed");
+      },
+    };
+    hub.subscribe(failing, { kind: "thread-detail", threadId: "thread-1" });
+    hub.subscribe(healthy, { kind: "thread-detail", threadId: "thread-1" });
+
+    expect(() =>
+      hub.notifyThread("thread-1", ["status-changed"]),
+    ).not.toThrow();
+
+    expect(closed).toEqual([{ code: 1013, reason: "app-send-failed" }]);
+    expect(healthy.messages).toHaveLength(1);
+
+    // The failing socket was unregistered; the next notify skips it.
+    hub.notifyThread("thread-1", ["status-changed"]);
+    expect(closed).toHaveLength(1);
+    expect(healthy.messages).toHaveLength(2);
+  });
+
   it("notifies all clients subscribed to the same thread", () => {
     const hub = new NotificationHub();
     const socket1 = createMockHubSocket();

From ab07364248260981bc3221e2709c1c1caea94ede Mon Sep 17 00:00:00 2001
From: Vedran Burojevic <vedran.burojevic@gmail.com>
Date: Mon, 24 Aug 2026 17:34:47 +0200
Subject: [PATCH 4/5] Single-serialize, rate-limit, and cap
 bb.realtime.publish; batch bb.log

bb.realtime.publish serialized every payload three times on the event
loop (stringify to check serializability, parse back, zod-parse the
envelope, stringify the frame), had no size bound, and no rate bound -
a runaway in-process publisher multiplied by every connected client.
bb.log did mkdirSync + statSync + appendFileSync per line on the loop.

publish now serializes the payload exactly once; that string is the
size-cap measurement (64KB, rejected with a clear error) and the exact
bytes on the wire - the hub validates the envelope fields through the
strict outgoing schema and embeds the payload verbatim instead of
re-parsing it. A per-plugin token bucket (10/s, burst 20) drops signals
over the limit - they are ephemeral by contract - warning once per 10s
window with the plugin id.

bb.log now buffers lines and flushes them as one async appendFile per
200ms window or 8KB, creates the log directory once per writer, and
tracks rotation from a cached size counter seeded by a single stat. The
tail path flushes before reading so it stays read-your-writes, and the
plugin API handle flushes and drops the writer in its dispose hooks so
a pending flush timer cannot race a plugin data-dir removal.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .../server/src/services/plugins/plugin-api.ts |  76 ++++++++-
 .../server/src/services/plugins/plugin-log.ts | 155 ++++++++++++++++--
 apps/server/src/ws/hub.ts                     |  30 +++-
 .../server/test/app/hub-plugin-signal.test.ts |  10 +-
 apps/server/test/app/hub.test.ts              |   8 +-
 .../test/services/plugins/plugin-cli.test.ts  |  24 +--
 .../test/services/plugins/plugin-log.test.ts  | 108 ++++++++++++
 .../test/services/plugins/plugin-wire.test.ts |  61 +++++++
 8 files changed, 421 insertions(+), 51 deletions(-)
 create mode 100644 apps/server/test/services/plugins/plugin-log.test.ts

diff --git a/apps/server/src/services/plugins/plugin-api.ts b/apps/server/src/services/plugins/plugin-api.ts
index 68d2c57dbc..927006230a 100644
--- a/apps/server/src/services/plugins/plugin-api.ts
+++ b/apps/server/src/services/plugins/plugin-api.ts
@@ -92,7 +92,7 @@ import type {
 import type { BbSdk, ThreadForkArgs, ThreadSpawnArgs } from "@bb/sdk";
 import type { ServerLogger } from "../../types.js";
 import type { PluginInteractionResult } from "../interactions/pending-interactions.js";
-import { appendPluginLogLine } from "./plugin-log.js";
+import { appendPluginLogLine, disposePluginLogWriter } from "./plugin-log.js";
 import type { PluginHostArtifactSnapshot } from "./plugin-service-internal.js";
 import { readPluginSettingsValues } from "./plugin-settings.js";
 
@@ -430,6 +430,19 @@ function createStagedRegistrations<
   };
 }
 
+/**
+ * `bb.realtime.publish` broadcasts to every connected client with no
+ * per-channel subscription, so a runaway publisher multiplies across the
+ * whole socket fan-out. Interactive signals are a few per second; anything
+ * sustained above that is treated as a loop and dropped (signals are
+ * ephemeral by contract), warning once per window so the log names the
+ * plugin without flooding.
+ */
+const REALTIME_PUBLISH_MAX_PAYLOAD_BYTES = 64 * 1024;
+const REALTIME_PUBLISH_TOKENS_PER_SECOND = 10;
+const REALTIME_PUBLISH_BURST_TOKENS = 20;
+const REALTIME_PUBLISH_DROP_WARN_INTERVAL_MS = 10_000;
+
 export function createPluginApi(options: {
   pluginId: string;
   logger: ServerLogger;
@@ -439,8 +452,10 @@ export function createPluginApi(options: {
   getSdk: () => BbSdk | undefined;
   /** Undefined until the server is listening (bb.server is bind-gated too). */
   getLoopbackBaseUrl: () => string | undefined;
-  /** Broadcasts a plugin-signal WS message (hub.notifyPluginSignal). */
-  publishSignal: (channel: string, payload: unknown) => void;
+  /** Broadcasts a plugin-signal WS message (hub.notifyPluginSignal). The
+   * payload arrives pre-serialized — publish's one JSON.stringify — and is
+   * embedded verbatim in the wire frame. */
+  publishSignal: (channel: string, serializedPayload: string) => void;
   /** Marks the plugin needs-configuration in the loader's status table. */
   reportNeedsConfiguration: (message: string) => void;
   /** Returns the owning plugin id when another plugin already registered
@@ -554,6 +569,10 @@ export function createPluginApi(options: {
   const pendingAgentToolProblems: string[] = [];
   const pendingSharedPorts = new Map<string, readonly number[]>();
   const disposeHooks: Array<() => void | Promise<void>> = [];
+  // Flush buffered bb.log lines before this generation's resources go away:
+  // a pending flush timer must not outlive (and race) a plugin data-dir
+  // removal.
+  disposeHooks.push(() => disposePluginLogWriter(dataDir, pluginId));
   const settingsRecord: PluginApiHandle["settings"] = {
     descriptors: {},
     listeners: [],
@@ -866,16 +885,45 @@ export function createPluginApi(options: {
     },
   };
 
+  // Token bucket for realtime publishes; state lives with the handle, so a
+  // reload starts the plugin with a full burst again.
+  let realtimePublishTokens = REALTIME_PUBLISH_BURST_TOKENS;
+  let realtimePublishRefilledAt = Date.now();
+  let realtimePublishWarnedAt = 0;
+  function takeRealtimePublishToken(channel: string): boolean {
+    const now = Date.now();
+    realtimePublishTokens = Math.min(
+      REALTIME_PUBLISH_BURST_TOKENS,
+      realtimePublishTokens +
+        ((now - realtimePublishRefilledAt) / 1000) *
+          REALTIME_PUBLISH_TOKENS_PER_SECOND,
+    );
+    realtimePublishRefilledAt = now;
+    if (realtimePublishTokens >= 1) {
+      realtimePublishTokens -= 1;
+      return true;
+    }
+    if (now - realtimePublishWarnedAt >= REALTIME_PUBLISH_DROP_WARN_INTERVAL_MS) {
+      realtimePublishWarnedAt = now;
+      logger.warn(
+        { pluginId, channel },
+        "Dropping bb.realtime.publish signals over the per-plugin rate limit",
+      );
+    }
+    return false;
+  }
+
   const realtime: PluginRealtime = {
     publish(channel, payload) {
       assertLive();
       if (typeof channel !== "string" || channel.length === 0) {
         throw new Error("realtime channel must be a non-empty string");
       }
-      // JSON round-trip up front: enforces serializability with a clear
-      // error at the publish site and strips prototypes/getters before the
-      // payload crosses the WS boundary.
-      let normalized: unknown = null;
+      // One serialization up front: it enforces serializability with a clear
+      // error at the publish site AND produces the exact bytes that cross
+      // the WS boundary — the hub embeds this string verbatim instead of
+      // re-parsing and re-stringifying the payload per publish.
+      let payloadJson = "null";
       if (payload !== undefined) {
         let json: string | undefined;
         try {
@@ -888,9 +936,19 @@ export function createPluginApi(options: {
             `realtime payload for channel "${channel}" is not JSON-serializable`,
           );
         }
-        normalized = JSON.parse(json);
+        payloadJson = json;
+      }
+      const payloadBytes = Buffer.byteLength(payloadJson, "utf8");
+      if (payloadBytes > REALTIME_PUBLISH_MAX_PAYLOAD_BYTES) {
+        throw new Error(
+          `realtime payload for channel "${channel}" is ${payloadBytes} bytes; ` +
+            `the limit is ${REALTIME_PUBLISH_MAX_PAYLOAD_BYTES} (64KB)`,
+        );
+      }
+      if (!takeRealtimePublishToken(channel)) {
+        return;
       }
-      publishSignal(channel, normalized);
+      publishSignal(channel, payloadJson);
     },
   };
 
diff --git a/apps/server/src/services/plugins/plugin-log.ts b/apps/server/src/services/plugins/plugin-log.ts
index 06afbf4a51..4ea6f0b3b0 100644
--- a/apps/server/src/services/plugins/plugin-log.ts
+++ b/apps/server/src/services/plugins/plugin-log.ts
@@ -1,5 +1,5 @@
-import { appendFileSync, mkdirSync, renameSync, statSync } from "node:fs";
-import { readFile } from "node:fs/promises";
+import { mkdirSync, renameSync, statSync } from "node:fs";
+import { appendFile, readFile } from "node:fs/promises";
 import { join } from "node:path";
 
 /**
@@ -7,21 +7,134 @@ import { join } from "node:path";
  * appended as JSONL to <dataDir>/plugins/<id>/logs/plugin.log in addition to
  * the prefixed server log. Simple size rotation: past 5MB the file is renamed
  * to plugin.log.1 (one rotated file kept, replacing the previous one).
+ *
+ * `bb.log` runs on the server event loop, so lines are buffered and flushed
+ * as one async append per window (or once the buffer crosses a byte
+ * threshold) instead of a mkdir + stat + sync append per line. The directory
+ * is created once per writer lifetime and the rotation size is tracked from
+ * a cached counter seeded by one stat. Delivery stays best effort — it
+ * already was, and the prefixed server log still carries every message.
  */
 const PLUGIN_LOG_MAX_BYTES = 5 * 1024 * 1024;
 const PLUGIN_LOG_FILE = "plugin.log";
 const PLUGIN_LOG_ROTATED_FILE = "plugin.log.1";
+const PLUGIN_LOG_FLUSH_INTERVAL_MS = 200;
+const PLUGIN_LOG_FLUSH_THRESHOLD_BYTES = 8 * 1024;
 
 type PluginLogLevel = "debug" | "info" | "warn" | "error";
 
+interface PluginLogWriter {
+  bufferedBytes: number;
+  bufferedLines: string[];
+  dirEnsured: boolean;
+  /** Serializes flushes so appended lines keep their order. */
+  flushChain: Promise<void>;
+  /** Cached plugin.log size; null until seeded by the writer's one stat. */
+  logFileBytes: number | null;
+  timer: ReturnType<typeof setTimeout> | null;
+}
+
+const pluginLogWriters = new Map<string, PluginLogWriter>();
+
 function pluginLogsDir(dataDir: string, pluginId: string): string {
   return join(dataDir, "plugins", pluginId, "logs");
 }
 
+function pluginLogWriterKey(dataDir: string, pluginId: string): string {
+  return `${dataDir}\u0000${pluginId}`;
+}
+
+function getPluginLogWriter(dataDir: string, pluginId: string): PluginLogWriter {
+  const key = pluginLogWriterKey(dataDir, pluginId);
+  const existing = pluginLogWriters.get(key);
+  if (existing !== undefined) {
+    return existing;
+  }
+  const writer: PluginLogWriter = {
+    bufferedBytes: 0,
+    bufferedLines: [],
+    dirEnsured: false,
+    flushChain: Promise.resolve(),
+    logFileBytes: null,
+    timer: null,
+  };
+  pluginLogWriters.set(key, writer);
+  return writer;
+}
+
+function flushPluginLogWriter(
+  dataDir: string,
+  pluginId: string,
+  writer: PluginLogWriter,
+): Promise<void> {
+  if (writer.timer !== null) {
+    clearTimeout(writer.timer);
+    writer.timer = null;
+  }
+  if (writer.bufferedLines.length === 0) {
+    return writer.flushChain;
+  }
+  const chunk = writer.bufferedLines.join("");
+  const chunkBytes = writer.bufferedBytes;
+  writer.bufferedLines = [];
+  writer.bufferedBytes = 0;
+  writer.flushChain = writer.flushChain.then(async () => {
+    try {
+      const dir = pluginLogsDir(dataDir, pluginId);
+      if (!writer.dirEnsured) {
+        mkdirSync(dir, { recursive: true });
+        writer.dirEnsured = true;
+      }
+      const file = join(dir, PLUGIN_LOG_FILE);
+      if (writer.logFileBytes === null) {
+        try {
+          writer.logFileBytes = statSync(file).size;
+        } catch {
+          // Missing file: nothing logged there yet.
+          writer.logFileBytes = 0;
+        }
+      }
+      if (writer.logFileBytes > PLUGIN_LOG_MAX_BYTES) {
+        try {
+          renameSync(file, join(dir, PLUGIN_LOG_ROTATED_FILE));
+        } catch {
+          // Missing file: nothing to rotate.
+        }
+        writer.logFileBytes = 0;
+      }
+      await appendFile(file, chunk, "utf8");
+      writer.logFileBytes += chunkBytes;
+    } catch {
+      // Best effort only — a full disk or permission problem must not break
+      // the plugin call site; the prefixed server log still carries the
+      // messages.
+    }
+  });
+  return writer.flushChain;
+}
+
+/**
+ * Flush and drop a plugin's buffered log writer. Wired into the plugin API
+ * handle's dispose hooks so reload/disable/shutdown wait for pending lines
+ * instead of leaving a flush timer racing the plugin data directory's
+ * removal.
+ */
+export async function disposePluginLogWriter(
+  dataDir: string,
+  pluginId: string,
+): Promise<void> {
+  const key = pluginLogWriterKey(dataDir, pluginId);
+  const writer = pluginLogWriters.get(key);
+  if (writer === undefined) {
+    return;
+  }
+  pluginLogWriters.delete(key);
+  await flushPluginLogWriter(dataDir, pluginId, writer);
+}
+
 /**
- * Append one log line synchronously (bb.log is a sync API; lines are tiny).
- * Never throws — a full disk or permission problem must not break the plugin
- * call site; the prefixed server log still carries the message.
+ * Buffer one log line (bb.log is a sync API; lines are tiny). Never throws —
+ * delivery is best effort and happens off the call path.
  */
 export function appendPluginLogLine(
   dataDir: string,
@@ -30,18 +143,21 @@ export function appendPluginLogLine(
   message: string,
 ): void {
   try {
-    const dir = pluginLogsDir(dataDir, pluginId);
-    mkdirSync(dir, { recursive: true });
-    const file = join(dir, PLUGIN_LOG_FILE);
-    try {
-      if (statSync(file).size > PLUGIN_LOG_MAX_BYTES) {
-        renameSync(file, join(dir, PLUGIN_LOG_ROTATED_FILE));
-      }
-    } catch {
-      // Missing file: nothing to rotate.
+    const writer = getPluginLogWriter(dataDir, pluginId);
+    const line = `${JSON.stringify({ ts: Date.now(), level, message })}\n`;
+    writer.bufferedLines.push(line);
+    writer.bufferedBytes += Buffer.byteLength(line, "utf8");
+    if (writer.bufferedBytes >= PLUGIN_LOG_FLUSH_THRESHOLD_BYTES) {
+      void flushPluginLogWriter(dataDir, pluginId, writer);
+      return;
+    }
+    if (writer.timer === null) {
+      writer.timer = setTimeout(() => {
+        writer.timer = null;
+        void flushPluginLogWriter(dataDir, pluginId, writer);
+      }, PLUGIN_LOG_FLUSH_INTERVAL_MS);
+      writer.timer.unref?.();
     }
-    const line = JSON.stringify({ ts: Date.now(), level, message });
-    appendFileSync(file, `${line}\n`, "utf8");
   } catch {
     // Best effort only.
   }
@@ -53,13 +169,18 @@ function splitLines(content: string): string[] {
 
 /**
  * Last `tail` log lines across the rotated file plus the current one, oldest
- * first. Missing files read as empty.
+ * first. Missing files read as empty. Pending buffered lines are flushed
+ * first so a tail right after `bb.log` stays read-your-writes.
  */
 export async function readPluginLogTail(
   dataDir: string,
   pluginId: string,
   tail: number,
 ): Promise<string[]> {
+  const writer = pluginLogWriters.get(pluginLogWriterKey(dataDir, pluginId));
+  if (writer !== undefined) {
+    await flushPluginLogWriter(dataDir, pluginId, writer);
+  }
   const dir = pluginLogsDir(dataDir, pluginId);
   const lines: string[] = [];
   for (const name of [PLUGIN_LOG_ROTATED_FILE, PLUGIN_LOG_FILE]) {
diff --git a/apps/server/src/ws/hub.ts b/apps/server/src/ws/hub.ts
index ffbf212508..fb652f8c27 100644
--- a/apps/server/src/ws/hub.ts
+++ b/apps/server/src/ws/hub.ts
@@ -59,6 +59,13 @@ const THREAD_LIST_EVENTS_APPENDED_COALESCE_MS = 1_000;
 const LIST_RELEVANT_THREAD_EVENT_TYPES: ReadonlySet<ThreadEventType> =
   new Set<ThreadEventType>(["client/turn/requested", "turn/completed"]);
 
+/**
+ * The plugin-signal envelope minus the payload: `notifyPluginSignal` embeds
+ * the publish site's already-serialized payload verbatim, so only the fields
+ * it splices into the frame go through the outgoing schema.
+ */
+const pluginSignalEnvelopeSchema = pluginSignalSchema.omit({ payload: true });
+
 interface HubSocket {
   close(code?: number, reason?: string): void;
   raw?: { bufferedAmount: number };
@@ -895,20 +902,25 @@ export class NotificationHub implements DbNotifier {
    * every connected client. V1 broadcasts to all clients — per-channel
    * subscriptions arrive with the plugin frontend runtime. Returns how many
    * clients the signal reached.
+   *
+   * `serializedPayload` is the publish site's single JSON serialization of
+   * the payload, embedded verbatim in the wire frame so the fan-out never
+   * parses or re-stringifies it. The envelope fields still pass the strict
+   * outgoing schema.
    */
   notifyPluginSignal(
     pluginId: string,
     channel: string,
-    payload: unknown,
+    serializedPayload: string,
   ): number {
-    const message = JSON.stringify(
-      pluginSignalSchema.parse({
-        type: "plugin-signal",
-        pluginId,
-        channel,
-        payload,
-      }),
-    );
+    const envelope = pluginSignalEnvelopeSchema.parse({
+      type: "plugin-signal",
+      pluginId,
+      channel,
+    });
+    const message = `{"type":"plugin-signal","pluginId":${JSON.stringify(
+      envelope.pluginId,
+    )},"channel":${JSON.stringify(envelope.channel)},"payload":${serializedPayload}}`;
     return this.broadcastToAppSockets(message);
   }
 
diff --git a/apps/server/test/app/hub-plugin-signal.test.ts b/apps/server/test/app/hub-plugin-signal.test.ts
index 3d7e07e881..27a16263ca 100644
--- a/apps/server/test/app/hub-plugin-signal.test.ts
+++ b/apps/server/test/app/hub-plugin-signal.test.ts
@@ -12,9 +12,13 @@ describe("NotificationHub.notifyPluginSignal", () => {
     hub.subscribe(first, { kind: "thread-detail", threadId: "thr_1" });
     hub.subscribe(second, { kind: "system" });
 
-    const delivered = hub.notifyPluginSignal("linear", "issues-updated", {
-      count: 42,
-    });
+    // The payload arrives pre-serialized (the publish site's single
+    // JSON.stringify) and is embedded verbatim in the wire frame.
+    const delivered = hub.notifyPluginSignal(
+      "linear",
+      "issues-updated",
+      JSON.stringify({ count: 42 }),
+    );
 
     expect(delivered).toBe(2);
     for (const socket of [first, second]) {
diff --git a/apps/server/test/app/hub.test.ts b/apps/server/test/app/hub.test.ts
index 4dfafdfd2e..5c5eeab901 100644
--- a/apps/server/test/app/hub.test.ts
+++ b/apps/server/test/app/hub.test.ts
@@ -280,7 +280,7 @@ describe("NotificationHub", () => {
     };
     hub.registerClient(healthy);
     hub.registerClient(wedged);
-    const bigPayload = "x".repeat(1024 * 1024);
+    const bigPayload = JSON.stringify("x".repeat(1024 * 1024));
 
     for (let index = 0; index < 5; index += 1) {
       hub.notifyPluginSignal("plugin-1", "channel-1", bigPayload);
@@ -294,7 +294,11 @@ describe("NotificationHub", () => {
     expect(healthy.messages).toHaveLength(5);
 
     // The drop unregistered the socket: later broadcasts skip it entirely.
-    const delivered = hub.notifyPluginSignal("plugin-1", "channel-1", "after");
+    const delivered = hub.notifyPluginSignal(
+      "plugin-1",
+      "channel-1",
+      JSON.stringify("after"),
+    );
     expect(delivered).toBe(1);
     expect(wedged.closed).toHaveLength(1);
   });
diff --git a/apps/server/test/services/plugins/plugin-cli.test.ts b/apps/server/test/services/plugins/plugin-cli.test.ts
index 424026a514..5cb9a8c9a8 100644
--- a/apps/server/test/services/plugins/plugin-cli.test.ts
+++ b/apps/server/test/services/plugins/plugin-cli.test.ts
@@ -365,6 +365,19 @@ describe("plugin CLI commands (bb.cli.register + endpoints + skill + logs)", ()
   });
 
   it("bb.log writes JSONL to the plugin log file and the tail endpoint serves it", async () => {
+    // The tail endpoint flushes buffered lines (bb.log batches writes off the
+    // call path), so hit it before reading the raw file.
+    const response = await harness.app.request(
+      `${BASE}/api/v1/plugins/acme/logs?tail=1`,
+    );
+    expect(response.status).toBe(200);
+    const body = (await response.json()) as { ok: boolean; lines: string[] };
+    expect(body.ok).toBe(true);
+    expect(body.lines).toHaveLength(1);
+    expect(JSON.parse(body.lines[0] ?? "")).toMatchObject({
+      message: "cli plugin loaded",
+    });
+
     const logFile = join(
       harness.config.dataDir,
       "plugins",
@@ -383,17 +396,6 @@ describe("plugin CLI commands (bb.cli.register + endpoints + skill + logs)", ()
     expect(parsed.message).toBe("cli plugin loaded");
     expect(typeof parsed.ts).toBe("number");
 
-    const response = await harness.app.request(
-      `${BASE}/api/v1/plugins/acme/logs?tail=1`,
-    );
-    expect(response.status).toBe(200);
-    const body = (await response.json()) as { ok: boolean; lines: string[] };
-    expect(body.ok).toBe(true);
-    expect(body.lines).toHaveLength(1);
-    expect(JSON.parse(body.lines[0] ?? "")).toMatchObject({
-      message: "cli plugin loaded",
-    });
-
     const missing = await harness.app.request(
       `${BASE}/api/v1/plugins/nope/logs`,
     );
diff --git a/apps/server/test/services/plugins/plugin-log.test.ts b/apps/server/test/services/plugins/plugin-log.test.ts
new file mode 100644
index 0000000000..af76a88d76
--- /dev/null
+++ b/apps/server/test/services/plugins/plugin-log.test.ts
@@ -0,0 +1,108 @@
+import { existsSync, readFileSync } from "node:fs";
+import { mkdtemp, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { describe, expect, it, vi } from "vitest";
+import {
+  appendPluginLogLine,
+  readPluginLogTail,
+} from "../../../src/services/plugins/plugin-log.js";
+
+function logFilePath(dataDir: string, pluginId: string): string {
+  return join(dataDir, "plugins", pluginId, "logs", "plugin.log");
+}
+
+function fileLines(path: string): string[] {
+  return readFileSync(path, "utf8")
+    .split("\n")
+    .filter((line) => line.length > 0);
+}
+
+async function withDataDir(
+  run: (dataDir: string) => Promise<void>,
+): Promise<void> {
+  const dataDir = await mkdtemp(join(tmpdir(), "bb-plugin-log-"));
+  try {
+    await run(dataDir);
+  } finally {
+    await rm(dataDir, { recursive: true, force: true });
+  }
+}
+
+describe("plugin log batching", () => {
+  it("buffers lines off the call path and flushes them as one batch", async () => {
+    await withDataDir(async (dataDir) => {
+      for (let index = 0; index < 3; index += 1) {
+        appendPluginLogLine(dataDir, "batch", "info", `line-${index}`);
+      }
+      // Nothing hits the disk synchronously on the call path.
+      expect(existsSync(logFilePath(dataDir, "batch"))).toBe(false);
+
+      // The tail flushes pending lines first, so reads stay read-your-writes.
+      const lines = await readPluginLogTail(dataDir, "batch", 10);
+      expect(lines).toHaveLength(3);
+      expect(JSON.parse(lines[0] ?? "")).toMatchObject({
+        level: "info",
+        message: "line-0",
+      });
+      expect(fileLines(logFilePath(dataDir, "batch"))).toHaveLength(3);
+    });
+  });
+
+  it("flushes on the timer without a reader", async () => {
+    await withDataDir(async (dataDir) => {
+      appendPluginLogLine(dataDir, "timer", "warn", "solo");
+      expect(existsSync(logFilePath(dataDir, "timer"))).toBe(false);
+
+      await vi.waitFor(
+        () => {
+          expect(existsSync(logFilePath(dataDir, "timer"))).toBe(true);
+        },
+        { timeout: 2_000 },
+      );
+      expect(fileLines(logFilePath(dataDir, "timer"))).toHaveLength(1);
+    });
+  });
+
+  it("flushes immediately once the buffer crosses the byte threshold", async () => {
+    await withDataDir(async (dataDir) => {
+      const big = "x".repeat(4 * 1024);
+      appendPluginLogLine(dataDir, "burst", "info", big);
+      expect(existsSync(logFilePath(dataDir, "burst"))).toBe(false);
+
+      // The second line crosses the 8KB threshold: no timer wait.
+      appendPluginLogLine(dataDir, "burst", "info", big);
+      await vi.waitFor(
+        () => {
+          expect(existsSync(logFilePath(dataDir, "burst"))).toBe(true);
+        },
+        { timeout: 500 },
+      );
+      expect(fileLines(logFilePath(dataDir, "burst"))).toHaveLength(2);
+    });
+  });
+
+  it("rotates past the size cap using the cached size counter", async () => {
+    await withDataDir(async (dataDir) => {
+      appendPluginLogLine(
+        dataDir,
+        "rotate",
+        "info",
+        "x".repeat(6 * 1024 * 1024),
+      );
+      await readPluginLogTail(dataDir, "rotate", 1);
+
+      appendPluginLogLine(dataDir, "rotate", "info", "after-rotation");
+      const lines = await readPluginLogTail(dataDir, "rotate", 10);
+
+      const dir = join(dataDir, "plugins", "rotate", "logs");
+      expect(existsSync(join(dir, "plugin.log.1"))).toBe(true);
+      expect(fileLines(logFilePath(dataDir, "rotate"))).toHaveLength(1);
+      // The tail spans the rotated file plus the current one.
+      expect(lines).toHaveLength(2);
+      expect(JSON.parse(lines[1] ?? "")).toMatchObject({
+        message: "after-rotation",
+      });
+    });
+  });
+});
diff --git a/apps/server/test/services/plugins/plugin-wire.test.ts b/apps/server/test/services/plugins/plugin-wire.test.ts
index ff7879dc35..5cbcc29ebd 100644
--- a/apps/server/test/services/plugins/plugin-wire.test.ts
+++ b/apps/server/test/services/plugins/plugin-wire.test.ts
@@ -25,6 +25,11 @@ const WIRE_SOURCE = `
       input: z.object({ channel: z.string(), payload: z.unknown() }),
       output: z.literal("published"),
     },
+    publishMany: {
+      input: z.object({ channel: z.string(), count: z.number() }),
+      output: z.literal("published-many"),
+    },
+    publishHuge: { input: z.record(z.string(), z.unknown()), output: z.null() },
     publishBad: { input: z.record(z.string(), z.unknown()), output: z.null() },
     invalidOutput: { input: z.null(), output: z.string() },
     bigintResult: { input: z.null(), output: z.any() },
@@ -94,6 +99,16 @@ const WIRE_SOURCE = `
         bb.realtime.publish(input.channel, input.payload);
         return "published";
       },
+      publishMany: async (input: any) => {
+        for (let index = 0; index < input.count; index += 1) {
+          bb.realtime.publish(input.channel, { index });
+        }
+        return "published-many";
+      },
+      publishHuge: async () => {
+        bb.realtime.publish("huge", "x".repeat(65 * 1024));
+        return null;
+      },
       publishBad: async () => {
         bb.realtime.publish("bad", { n: BigInt(1) });
       },
@@ -667,6 +682,52 @@ describe("plugin wire surfaces (http/rpc dispatcher + realtime)", () => {
     });
   });
 
+  it("bb.realtime.publish rejects payloads over the 64KB cap", async () => {
+    const socket = createMockHubSocket();
+    harness.hub.subscribe(socket, { kind: "system" });
+
+    const response = await rpc(harness, "publishHuge", {});
+    expect(response.status).toBe(500);
+    expect(await response.json()).toMatchObject({
+      ok: false,
+      error: {
+        code: "handler_error",
+        message: expect.stringContaining("64KB"),
+      },
+    });
+    expect(socket.messages).toHaveLength(0);
+  });
+
+  it("bb.realtime.publish drops signals over the per-plugin rate limit with one warning", async () => {
+    const socket = createMockHubSocket();
+    harness.hub.subscribe(socket, { kind: "system" });
+    const originalWarn = harness.deps.logger.warn;
+    const warn = vi.fn();
+    harness.deps.logger.warn = warn;
+    try {
+      // 40 synchronous publishes land inside one refill tick, so exactly the
+      // burst allowance goes out and the rest drop with a single warning.
+      const response = await rpc(harness, "publishMany", {
+        channel: "flood",
+        count: 40,
+      });
+      expect(response.status).toBe(200);
+      expect(await response.json()).toEqual({
+        ok: true,
+        result: "published-many",
+      });
+
+      expect(socket.messages).toHaveLength(20);
+      const rateLimitWarnings = warn.mock.calls.filter(([, message]) =>
+        String(message).includes("rate limit"),
+      );
+      expect(rateLimitWarnings).toHaveLength(1);
+      expect(rateLimitWarnings[0]?.[0]).toMatchObject({ pluginId: "wire" });
+    } finally {
+      harness.deps.logger.warn = originalWarn;
+    }
+  });
+
   it("rpc resolves the handler after the body arrives, so a reload during the body read never runs a stale handler", async () => {
     // The handler closes over its load generation: a binding resolved
     // before the body read (and invalidated by the mid-read reload) would

From fd437191b0c63931cc14dc52326b6af86daf59f3 Mon Sep 17 00:00:00 2001
From: Vedran Burojevic <vedran.burojevic@gmail.com>
Date: Mon, 24 Aug 2026 17:41:02 +0200
Subject: [PATCH 5/5] Cap thread lists at a 200-row default and signal
 truncation

GET /threads and the projects-with-threads bootstrap ran unbounded list
queries: the route accepted an optional limit nobody passed, and the
per-projects variant had no limit parameter at all, so every sidebar
bootstrap and list read scaled with a long-lived install's full thread
history.

listThreadsWithPendingInteractionState now defaults its limit to
DEFAULT_THREAD_LIST_LIMIT (200) and the ForProjects variant gains the
same defaulted limit across all listed projects; both order by the
active-sidebar order, so the cap keeps the most relevant rows. The
GET /threads route fetches one probe row past the effective limit and
reports truncation in an x-bb-thread-list-has-more response header -
the response body stays a bare array for compatibility - so clients can
page with limit/offset once they hit the cap.

Known behavior change, deliberate: a `bb thread list` or unpaged SDK
list call now returns at most 200 rows (the two in-repo programmatic
consumers already pass explicit limits and page).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 apps/server/src/routes/threads/base.ts        | 14 ++++++--
 .../test/public/public-thread-data.test.ts    | 27 +++++++++++++++
 packages/db/src/data/index.ts                 |  1 +
 packages/db/src/data/threads.ts               | 22 ++++++++++---
 packages/db/test/data/threads.test.ts         | 33 +++++++++++++++++++
 5 files changed, 91 insertions(+), 6 deletions(-)

diff --git a/apps/server/src/routes/threads/base.ts b/apps/server/src/routes/threads/base.ts
index d99182e871..4e73cb6127 100644
--- a/apps/server/src/routes/threads/base.ts
+++ b/apps/server/src/routes/threads/base.ts
@@ -1,4 +1,5 @@
 import {
+  DEFAULT_THREAD_LIST_LIMIT,
   THREAD_SEARCH_LIMIT_PER_GROUP_DEFAULT,
   THREAD_SEARCH_LIMIT_PER_GROUP_MAX,
   countNonDeletedAssignedChildThreads,
@@ -233,7 +234,13 @@ export function registerThreadBaseRoutes(app: Hono, deps: AppDeps): void {
     if (query.sectionId) {
       requireThreadSection(deps, query.sectionId);
     }
-    const threads = listThreadsWithPendingInteractionState(deps.db, {
+    // The list is capped (default DEFAULT_THREAD_LIST_LIMIT) so one
+    // long-lived install cannot make every list read scale with its full
+    // history. The response stays a bare array for compatibility, so the
+    // truncation signal rides a header: fetch one probe row past the cap and
+    // report whether it existed. Callers page with limit/offset.
+    const effectiveLimit = limit ?? DEFAULT_THREAD_LIST_LIMIT;
+    const threadRows = listThreadsWithPendingInteractionState(deps.db, {
       ...(query.projectId ? { projectId: query.projectId } : {}),
       ...(query.parentThreadId ? { parentThreadId: query.parentThreadId } : {}),
       ...(query.sourceThreadId ? { sourceThreadId: query.sourceThreadId } : {}),
@@ -246,9 +253,12 @@ export function registerThreadBaseRoutes(app: Hono, deps: AppDeps): void {
         query.archived === undefined ? undefined : query.archived === "true",
       hasParent:
         query.hasParent === undefined ? undefined : query.hasParent === "true",
-      ...(limit !== undefined ? { limit } : {}),
+      limit: effectiveLimit + 1,
       ...(offset !== undefined ? { offset } : {}),
     });
+    const hasMore = threadRows.length > effectiveLimit;
+    const threads = hasMore ? threadRows.slice(0, effectiveLimit) : threadRows;
+    context.header("x-bb-thread-list-has-more", hasMore ? "true" : "false");
     return context.json(
       toThreadListEntryResponses(deps, { threads }) satisfies ThreadListEntry[],
     );
diff --git a/apps/server/test/public/public-thread-data.test.ts b/apps/server/test/public/public-thread-data.test.ts
index 41f204bd43..d59e082f88 100644
--- a/apps/server/test/public/public-thread-data.test.ts
+++ b/apps/server/test/public/public-thread-data.test.ts
@@ -210,6 +210,33 @@ describe("public thread data routes", () => {
     });
   });
 
+  it("caps thread lists and reports truncation in the has-more header", async () => {
+    await withTestHarness(async (harness) => {
+      const { host } = seedHostSession(harness.deps);
+      const { project } = seedProjectWithSource(harness.deps, {
+        hostId: host.id,
+      });
+      for (let index = 0; index < 3; index += 1) {
+        seedThread(harness.deps, {
+          projectId: project.id,
+          title: `Thread ${index}`,
+        });
+      }
+
+      const capped = await harness.app.request("/api/v1/threads?limit=2");
+      expect(capped.status).toBe(200);
+      expect(capped.headers.get("x-bb-thread-list-has-more")).toBe("true");
+      expect(z.array(z.unknown()).parse(await readJson(capped))).toHaveLength(
+        2,
+      );
+
+      const full = await harness.app.request("/api/v1/threads");
+      expect(full.status).toBe(200);
+      expect(full.headers.get("x-bb-thread-list-has-more")).toBe("false");
+      expect(z.array(z.unknown()).parse(await readJson(full))).toHaveLength(3);
+    });
+  });
+
   it("updates visibility and requires includeHidden to list hidden threads", async () => {
     await withTestHarness(async (harness) => {
       const { host } = seedHostSession(harness.deps);
diff --git a/packages/db/src/data/index.ts b/packages/db/src/data/index.ts
index 4cadc15b1a..2ebef46318 100644
--- a/packages/db/src/data/index.ts
+++ b/packages/db/src/data/index.ts
@@ -55,6 +55,7 @@ export {
   createThread,
   countLiveThreadsInEnvironment,
   countNonDeletedAssignedChildThreads,
+  DEFAULT_THREAD_LIST_LIMIT,
   getThread,
   getThreadExecutionOverride,
   hasActiveThreadAttention,
diff --git a/packages/db/src/data/threads.ts b/packages/db/src/data/threads.ts
index b09c32fc75..3df5356b35 100644
--- a/packages/db/src/data/threads.ts
+++ b/packages/db/src/data/threads.ts
@@ -373,6 +373,15 @@ export function listThreadMentionRowsByIds(
     .all();
 }
 
+/**
+ * Default row cap for thread listings. The sidebar and every SDK/CLI list
+ * ride these queries; without a cap one long-lived install (thousands of
+ * archived-then-restored threads) makes every list read scale with history.
+ * Generous enough that hitting it means paging, not filtering — callers pass
+ * an explicit `limit` to page or to read more.
+ */
+export const DEFAULT_THREAD_LIST_LIMIT = 200;
+
 export interface ListThreadsOptions {
   projectId?: string;
   archived?: boolean;
@@ -400,6 +409,12 @@ type ThreadRow = typeof threads.$inferSelect;
 export interface ListThreadsForProjectsOptions {
   projectIds: readonly string[];
   archived?: boolean;
+  /**
+   * Row cap across all listed projects combined (the query orders by the
+   * active-thread sidebar order, so the cap keeps the most relevant rows).
+   * Defaults to {@link DEFAULT_THREAD_LIST_LIMIT}.
+   */
+  limit?: number;
 }
 
 export interface PinThreadArgs {
@@ -1204,10 +1219,8 @@ export function listThreadsWithPendingInteractionState(
   let query = threadWithPendingInteractionBaseQuery(db)
     .where(and(...buildListThreadsFilters(options)))
     .orderBy(...buildListThreadsOrderBy(options))
-    .$dynamic();
-  if (options.limit !== undefined) {
-    query = query.limit(options.limit);
-  }
+    .$dynamic()
+    .limit(options.limit ?? DEFAULT_THREAD_LIST_LIMIT);
   if (options.offset !== undefined) {
     query = query.offset(options.offset);
   }
@@ -1261,6 +1274,7 @@ export function listThreadsWithPendingInteractionStateForProjects(
   const rows = threadWithPendingInteractionBaseQuery(db)
     .where(and(...buildListThreadsForProjectsFilters(options)))
     .orderBy(...buildListThreadsForProjectsOrderBy(options))
+    .limit(options.limit ?? DEFAULT_THREAD_LIST_LIMIT)
     .all();
 
   return rows.map(toThreadWithPendingInteractionState);
diff --git a/packages/db/test/data/threads.test.ts b/packages/db/test/data/threads.test.ts
index 69349b53fd..5256021f2a 100644
--- a/packages/db/test/data/threads.test.ts
+++ b/packages/db/test/data/threads.test.ts
@@ -7,6 +7,7 @@ import {
   createThread,
   countLiveThreadsInEnvironment,
   countNonDeletedAssignedChildThreads,
+  DEFAULT_THREAD_LIST_LIMIT,
   getThread,
   getThreadExecutionOverride,
   hasActiveThreadAttention,
@@ -235,6 +236,38 @@ describe("threads", () => {
     ).toBe(2);
   });
 
+  it("caps thread listings at the default limit and honors explicit limits", () => {
+    const { db, project } = setup();
+    for (let index = 0; index < DEFAULT_THREAD_LIST_LIMIT + 1; index += 1) {
+      createThread(db, noopNotifier, {
+        projectId: project.id,
+        providerId: "codex",
+      });
+    }
+
+    expect(
+      listThreadsWithPendingInteractionState(db, { projectId: project.id }),
+    ).toHaveLength(DEFAULT_THREAD_LIST_LIMIT);
+    expect(
+      listThreadsWithPendingInteractionState(db, {
+        projectId: project.id,
+        limit: DEFAULT_THREAD_LIST_LIMIT + 1,
+      }),
+    ).toHaveLength(DEFAULT_THREAD_LIST_LIMIT + 1);
+
+    expect(
+      listThreadsWithPendingInteractionStateForProjects(db, {
+        projectIds: [project.id],
+      }),
+    ).toHaveLength(DEFAULT_THREAD_LIST_LIMIT);
+    expect(
+      listThreadsWithPendingInteractionStateForProjects(db, {
+        projectIds: [project.id],
+        limit: 5,
+      }),
+    ).toHaveLength(5);
+  });
+
   it("allows hidden threads to belong to sections", () => {
     const { db, project } = setup();
     const section = mustCreateThreadSection(db, "Work");