Unblock the server event loop: timeline build coalescing, deferred pruning, websocket backpressure

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);

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[],
     );

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 },

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);
     },
   };