TanStack · AlemTuzlak · Sep 21, 2026 · Sep 21, 2026 · Sep 21, 2026 · coderabbitai
diff --git a/.changeset/custom-event-immediate-flush.md b/.changeset/custom-event-immediate-flush.md
@@ -0,0 +1,6 @@
+---
+'@tanstack/ai': minor
+'@tanstack/ai-sandbox': patch
+---
+
+Flush CUSTOM events through the durability layer as soon as they are emitted, so progress events such as `compaction:started` reach the client at emit time. High-volume events (`process.stdout`, `process.stderr`, `sandbox.file`, `sandbox.file.diff`) still batch. Pass `{ batch: true }` on `emitCustomEvent` to opt an event into the batch.
diff --git a/docs/advanced/compaction.md b/docs/advanced/compaction.md
@@ -197,9 +197,10 @@ The token count is a rough `characters / 4` estimate. It is good enough to trigg
 
 After a compaction, the chat stream includes three CUSTOM events in order:
 `compaction:started`, `compaction:state`, then `compaction:ended`.
-`compaction:started` is sent before the strategy runs, so a slow
-`summarizeOldest` call still shows up as started on the client. The state and
-ended events follow when the strategy returns.
+`compaction:started` is sent before the strategy runs. Durability flushes each
+event as soon as it is emitted, so a slow `summarizeOldest` call still shows
+as started on the client while it runs. The state and ended events follow when
+the strategy returns.
 TanStack AI DevTools has a Compaction tab on the hook. Each compact shows:
 
 - started, state, and ended rows

diff --git a/docs/advanced/middleware.md b/docs/advanced/middleware.md
@@ -692,7 +692,7 @@ Every hook receives a `ChatMiddlewareContext` as its first argument. It provides
 | `chunkIndex` | `number` | Running count of chunks yielded |
 | `signal` | `AbortSignal \| undefined` | External abort signal |
 | `abort(reason?)` | `function` | Abort the run from within middleware |
-| `emitCustomEvent(name, value)` | `function` | Push a `CUSTOM` chunk onto the chat stream now. The engine yields it while the current hook is still running, including during `onConfig`. |
+| `emitCustomEvent(name, value, options?)` | `function` | Push a `CUSTOM` chunk onto the chat stream now. The engine yields it while the current hook is still running, including during `onConfig`. Durability flushes it immediately. Pass `{ batch: true }` to keep it in the durability batch. |
 | `context` | `TContext` | User-provided runtime context value |
 | `defer(promise)` | `function` | Register a non-blocking side-effect |
 
@@ -999,7 +999,7 @@ const progress: ChatMiddleware = {
 };
 ```
 
-The engine yields each `CUSTOM` chunk as soon as you call `emitCustomEvent`. If `RUN_STARTED` is not on the wire yet, the engine sends it first. Read these events on the client the same way as tool `emitCustomEvent` calls. See [Custom Events](../protocol/custom-events).
+The engine yields each `CUSTOM` chunk as soon as you call `emitCustomEvent`. If `RUN_STARTED` is not on the wire yet, the engine sends it first. Durability then flushes the event so the client can render a live indicator. Read these events on the client the same way as tool `emitCustomEvent` calls. See [Custom Events](../protocol/custom-events).
 
 ### Rate Limiting
 

diff --git a/docs/api/ai.md b/docs/api/ai.md
@@ -632,7 +632,11 @@ interface Tool<TContext = unknown> {
 ```typescript ignore
 type ToolExecutionContext<TContext = unknown> = {
   toolCallId?: string;
-  emitCustomEvent: (eventName: string, value: Record<string, any>) => void;
+  emitCustomEvent: (
+    eventName: string,
+    value: Record<string, any>,
+    options?: { batch?: boolean },
+  ) => void;
 } & (unknown extends TContext ? { context?: TContext } : { context: TContext });
 ```
 

diff --git a/docs/protocol/custom-events.md b/docs/protocol/custom-events.md
@@ -162,6 +162,23 @@ const progress: ChatMiddleware = {
 };
 ```
 
+With durability on the response, each of these events flushes as soon as it
+is emitted, so `prepare` reaches the client while `prepare()` is still
+running. High-volume names stay in the durability batch:
+`process.stdout`, `process.stderr`, `sandbox.file`, and `sandbox.file.diff`.
+Pass `{ batch: true }` to keep one of your own events in that batch:
+
+```ts
+import { type ChatMiddleware } from "@tanstack/ai";
+
+const noisy: ChatMiddleware = {
+  name: "noisy",
+  async onConfig(ctx) {
+    ctx.emitCustomEvent("my-app:ticks", { n: 1 }, { batch: true });
+  },
+};
+```
+
 These flow over the wire exactly like the built-in events: same `CUSTOM`
 chunk shape, same runtime behavior. But `'my-app:progress'` isn't one of the
 literal names in `KnownCustomEvent`, so it's intentionally absent from

diff --git a/docs/reference/interfaces/ChatMiddlewareContext.md b/docs/reference/interfaces/ChatMiddlewareContext.md
@@ -181,14 +181,15 @@ after the terminal hook (onFinish/onAbort/onError).
 ### emitCustomEvent
 
 ```ts
-emitCustomEvent: (name, value) => void;
+emitCustomEvent: (name, value, options?) => void;
 ```
 
 Defined in: [packages/ai/src/activities/chat/middleware/types.ts:221](https://github.com/TanStack/ai/blob/main/packages/ai/src/activities/chat/middleware/types.ts#L221)
 
 Push a `CUSTOM` chunk onto the chat stream immediately.
 The engine yields it as soon as it can (including while `onConfig`
-is still awaiting work such as a summarize call).
+is still awaiting work such as a summarize call). Durability then
+flushes the event on its own, unless you pass `{ batch: true }`.
 
 #### Parameters
 
@@ -200,6 +201,10 @@ is still awaiting work such as a summarize call).
 
 `Record`\<`string`, `any`\>
 
+##### options?
+
+`EmitCustomEventOptions`
+
 #### Returns
 
 `void`

diff --git a/docs/reference/type-aliases/ToolExecutionContext.md b/docs/reference/type-aliases/ToolExecutionContext.md
@@ -27,11 +27,13 @@ e.g. MCP `callTool` — should forward this to cancel in-flight work.
 ### emitCustomEvent
 
 ```ts
-emitCustomEvent: (eventName, value) => void;
+emitCustomEvent: (eventName, value, options?) => void;
 ```
 
 Emit a custom event during tool execution.
 Events are streamed to the client in real-time as AG-UI CUSTOM events.
+Durability flushes each event immediately. Pass `{ batch: true }` to keep
+the event in the durability batch.
 
 #### Parameters
 
@@ -47,6 +49,12 @@ Name of the custom event
 
 Event payload value
 
+##### options?
+
+`EmitCustomEventOptions`
+
+Pass `{ batch: true }` to keep this event in the durability batch
+
 #### Returns
 
 `void`

diff --git a/docs/tools/tools.md b/docs/tools/tools.md
@@ -431,6 +431,10 @@ const importData = importDataDef.server<ImportContext>(async (input, { context,
 });
 ```
 
+Each `emitCustomEvent` call flushes through durability immediately, so the
+client can show progress while the tool still runs. Pass `{ batch: true }`
+only for a high-volume stream. See [Custom Events](../protocol/custom-events).
+
 See [Server Tools](./server-tools) for the full runtime-context pattern.
 
 ## Tool States

diff --git a/packages/ai-sandbox/src/bridge-events.ts b/packages/ai-sandbox/src/bridge-events.ts
@@ -11,11 +11,15 @@
  * runs (e.g. code mode's `code_mode:console` logs during a long execution).
  */
 import { EventType, withTanstackMetadata } from '@tanstack/ai'
-import type { StreamChunk } from '@tanstack/ai'
+import type { EmitCustomEventOptions, StreamChunk } from '@tanstack/ai'
 
 export interface BridgeEventChannel {
   /** Pass as the bridge's `emitCustomEvent`; buffers a CUSTOM chunk for the stream. */
-  emitCustomEvent: (eventName: string, value: Record<string, unknown>) => void
+  emitCustomEvent: (
+    eventName: string,
+    value: Record<string, unknown>,
+    options?: EmitCustomEventOptions,
+  ) => void
   /** Live CUSTOM-chunk stream; ends after {@link close} once drained. */
   stream: AsyncIterable<StreamChunk>
   /** Stop the stream (call when the run's main output is done). */
@@ -48,7 +52,7 @@ export function createBridgeEventChannel(meta: {
   }
 
   return {
-    emitCustomEvent(eventName, value) {
+    emitCustomEvent(eventName, value, options) {
       if (closed) return
       buffer.push(
         withTanstackMetadata(
@@ -62,6 +66,7 @@ export function createBridgeEventChannel(meta: {
             model: meta.model,
             ...(meta.threadId !== undefined ? { threadId: meta.threadId } : {}),
             ...(meta.runId !== undefined ? { runId: meta.runId } : {}),
+            ...(options?.batch === true ? { batch: true } : {}),
           },
         ) as StreamChunk,
       )

diff --git a/packages/ai-sandbox/src/tool-bridge.ts b/packages/ai-sandbox/src/tool-bridge.ts
@@ -28,7 +28,7 @@ import {
   ListToolsRequestSchema,
 } from '@modelcontextprotocol/sdk/types.js'
 import type { AddressInfo } from 'node:net'
-import type { AnyTool } from '@tanstack/ai'
+import type { AnyTool, EmitCustomEventOptions } from '@tanstack/ai'
 
 /**
  * Name of the bridged MCP server. The agent sees tools as
@@ -72,7 +72,11 @@ export interface ToolBridgeCoreOptions {
    * `emitCustomEvent` never reaches a bridged tool. The harness adapter supplies
    * one that injects a CUSTOM chunk into its live output stream.
    */
-  emitCustomEvent?: (eventName: string, value: Record<string, unknown>) => void
+  emitCustomEvent?: (
+    eventName: string,
+    value: Record<string, unknown>,
+    options?: EmitCustomEventOptions,
+  ) => void
   /**
    * Optional permission-prompt tool (e.g. for Claude Code's
    * `--permission-prompt-tool`). When set, the bridge exposes an extra MCP tool

diff --git a/packages/ai/src/activities/chat/index.ts b/packages/ai/src/activities/chat/index.ts
@@ -38,6 +38,7 @@ import {
   tanstackMetadata,
   withTanstackMetadata,
 } from '../../utilities/merge-metadata'
+import { withDurabilityBatchHint } from '../../utilities/durability-batch'
 import { normalizeStreamChunk } from '../../utilities/normalize-stream-chunk'
 import { restorePublicUsage } from '../../utilities/restore-inbound-chunk'
 import type { AdapterYieldChunk } from '../../utilities/adapter-yield-chunk'
@@ -101,6 +102,7 @@ import type {
   ChatStream,
   ConstrainedModelMessage,
   CustomEvent,
+  EmitCustomEventOptions,
   InferSchemaType,
   Interrupt,
   JSONSchema,
@@ -1000,9 +1002,9 @@ class TextEngine<
         this.abortReason = reason
         this.middlewareAbortController?.abort(reason)
       },
-      emitCustomEvent: (name, value) => {
+      emitCustomEvent: (name, value, options) => {
         this.middlewareCustomQueue.push(
-          this.createCustomEventChunk(name, value),
+          this.createCustomEventChunk(name, value, options),
         )
         const waiters = this.middlewareCustomWaiters
         this.middlewareCustomWaiters = []
@@ -2019,7 +2021,8 @@ class TextEngine<
       this.resolveExecutableTools(executablePendingCalls),
       approvals,
       clientToolResults,
-      (eventName, data) => this.createCustomEventChunk(eventName, data),
+      (eventName, data, options) =>
+        this.createCustomEventChunk(eventName, data, options),
       {
         onBeforeToolCall: async (toolCall, tool, args) => {
           this.logger.tools(`phase=before name=${toolCall.function.name}`, {
@@ -2199,7 +2202,8 @@ class TextEngine<
       this.resolveExecutableTools(executableToolCalls),
       approvals,
       clientToolResults,
-      (eventName, data) => this.createCustomEventChunk(eventName, data),
+      (eventName, data, options) =>
+        this.createCustomEventChunk(eventName, data, options),
       {
         onBeforeToolCall: async (toolCall, tool, args) => {
           this.logger.tools(`phase=before name=${toolCall.function.name}`, {
@@ -4583,13 +4587,15 @@ class TextEngine<
   private createCustomEventChunk(
     eventName: string,
     value: Record<string, unknown>,
+    options?: EmitCustomEventOptions,
   ): CustomEvent {
-    return {
+    const chunk: CustomEvent = {
       type: EventType.CUSTOM,
       timestamp: Date.now(),
       name: eventName,
       value,
     }
+    return options?.batch ? withDurabilityBatchHint(chunk) : chunk
   }
 
   private createId(prefix: string): string {

diff --git a/packages/ai/src/activities/chat/middleware/types.ts b/packages/ai/src/activities/chat/middleware/types.ts
@@ -4,6 +4,7 @@ import type {
 } from '@standard-schema/spec'
 import type {
   AgentLoopState,
+  EmitCustomEventOptions,
   JSONSchema,
   ModelMessage,
   RunAgentResumeItem,
@@ -216,9 +217,14 @@ export interface ChatMiddlewareContext<TContext = unknown> {
   /**
    * Push a `CUSTOM` chunk onto the chat stream immediately.
    * The engine yields it as soon as it can (including while `onConfig`
-   * is still awaiting work such as a summarize call).
+   * is still awaiting work such as a summarize call). Durability then
+   * flushes the event on its own, unless you pass `{ batch: true }`.
    */
-  emitCustomEvent: (name: string, value: Record<string, any>) => void
+  emitCustomEvent: (
+    name: string,
+    value: Record<string, any>,
+    options?: EmitCustomEventOptions,
+  ) => void
   /** Runtime context provided by chat() options */
   context: TContext
   /**

diff --git a/packages/ai/src/activities/chat/tools/tool-calls.ts b/packages/ai/src/activities/chat/tools/tool-calls.ts
@@ -7,6 +7,7 @@ import type {
   AnyTool,
   ContentPart,
   CustomEvent,
+  EmitCustomEventOptions,
   ModelMessage,
   RunFinishedEvent,
   Tool,
@@ -760,6 +761,7 @@ export async function* executeToolCalls<TContext = unknown>(
   createCustomEventChunk?: (
     eventName: string,
     value: Record<string, any>,
+    options?: EmitCustomEventOptions,
   ) => CustomEvent,
   middlewareHooks?: ToolExecutionMiddlewareHooks,
   userContext?: TContext,
@@ -874,13 +876,21 @@ export async function* executeToolCalls<TContext = unknown>(
       toolCallId: toolCall.id,
       context: userContext,
       abortSignal,
-      emitCustomEvent: (eventName: string, value: Record<string, any>) => {
+      emitCustomEvent: (
+        eventName: string,
+        value: Record<string, any>,
+        options?: EmitCustomEventOptions,
+      ) => {
         if (createCustomEventChunk) {
           pendingEvents.push(
-            createCustomEventChunk(eventName, {
-              ...value,
-              toolCallId: toolCall.id,
-            }),
+            createCustomEventChunk(
+              eventName,
+              {
+                ...value,
+                toolCallId: toolCall.id,
+              },
+              options,
+            ),
           )
         }
       },

diff --git a/packages/ai/src/stream-to-response.ts b/packages/ai/src/stream-to-response.ts
@@ -9,6 +9,10 @@ import { notifyRunDisconnected } from './delivery-disconnect'
 import { resolveResumeRunId } from './stream-durability'
 import { EventType } from './types'
 import { toWireChunk } from './strip-to-spec-middleware'
+import {
+  isDurabilityBatchedCustom,
+  stripDurabilityBatchHint,
+} from './utilities/durability-batch'
 import { resolveDebugOption } from './logger/resolve'
 import { runErrorEventToError } from './utilities/errors'
 import type { LockStore } from './activities/chat/middleware/locks'
@@ -319,23 +323,29 @@ function resolveBatchSize(batch: number | undefined): number {
 
 /**
  * Boundaries at which the batching producer flushes early, regardless of the
- * batch size — the run-start marker, terminal events, and tool-call ends.
- * Flushing here keeps the durability log promptly consistent at semantically
- * meaningful points.
+ * batch size: run-start, terminals, tool-call ends, and CUSTOM events that
+ * are not high-volume adapter output.
  *
  * `RUN_STARTED` matters especially for one-shot activities (image, speech,
  * transcription, summarize): they emit `RUN_STARTED`, then await the provider
  * for seconds, then a terminal. Without flushing `RUN_STARTED` the log stays
  * empty for the whole run, so a mount-time `joinRun` finds nothing and its
- * empty-log deadline fast-fails as "run gone" — even though the run is alive.
+ * empty-log deadline fast-fails as "run gone" even though the run is alive.
  * Flushing it immediately makes the run resumable from the instant it starts.
+ *
+ * CUSTOM progress events (compaction, tool progress, middleware) flush at
+ * emit time so a live indicator can render. `process.stdout`,
+ * `process.stderr`, `sandbox.file`, and `sandbox.file.diff` stay batched.
+ * `emitCustomEvent(name, value, { batch: true })` opts a single event into
+ * that same batch.
  */
 function isDurabilityFlushBoundary(chunk: StreamChunk): boolean {
   return (
     chunk.type === 'RUN_STARTED' ||
     chunk.type === 'RUN_FINISHED' ||
     chunk.type === 'RUN_ERROR' ||
-    chunk.type === 'TOOL_CALL_END'
+    chunk.type === 'TOOL_CALL_END' ||
+    (chunk.type === 'CUSTOM' && !isDurabilityBatchedCustom(chunk))
   )
 }
 
@@ -446,7 +456,7 @@ export function durableStreamSource<TOffset extends string>(
 
     async function* flush(): AsyncIterable<StreamChunk> {
       if (batch.length === 0) return
-      const toForward = batch
+      const toForward = batch.map(stripDurabilityBatchHint)
       batch = []
       // Tag each chunk with the exact backend offset. Requiring one opaque
       // token per chunk preserves exact-once resume at any batch size.