feat(sdk,core): ChatChunkTooLargeError for oversized chat-stream chunks

ericallam · ericallam · commit bc751bd73854 · 2026-04-30T21:03:25.000+01:00
The realtime stream caps each record at ~1 MiB. Today the chat.agent path
through StreamsWriterV2 surfaces a generic S2Error from deep in the
batching layer when a chunk exceeds the cap, with no chunk-type context
and no guidance for callers.

Add a pre-write byte check in StreamsWriterV2.initializeServerStream that
fires before the chunk hits the underlying batcher, and a typed
ChatChunkTooLargeError carrying the chunk's discriminant (type/kind),
serialized size, and cap. Also exports an isChatChunkTooLargeError guard
from the SDK so callers can branch cleanly.

Threshold is 1 MiB minus 1 KiB to leave headroom for the JSON record
envelope. The error message links to the new docs pattern (Pattern:
ID-reference for large tool outputs / out-of-band streams.writer for
run-scoped data).
diff --git a/packages/core/src/v3/errors.ts b/packages/core/src/v3/errors.ts
@@ -629,6 +629,26 @@ export class GracefulExitTimeoutError extends Error {
   }
 }
 
+export class ChatChunkTooLargeError extends Error {
+  constructor(
+    public readonly chunkSize: number,
+    public readonly maxSize: number,
+    public readonly chunkType?: string
+  ) {
+    super(
+      `chat.agent chunk${chunkType ? ` of type "${chunkType}"` : ""} is ${chunkSize} bytes, ` +
+        `over the realtime stream's per-record cap of ${maxSize} bytes. ` +
+        `For oversized payloads (e.g. large tool outputs), write the value to your own store and ` +
+        `emit only an id/url through the chat stream — see https://trigger.dev/docs/ai-chat/patterns/large-payloads.`
+    );
+    this.name = "ChatChunkTooLargeError";
+  }
+}
+
+export function isChatChunkTooLargeError(error: unknown): error is ChatChunkTooLargeError {
+  return error instanceof Error && error.name === "ChatChunkTooLargeError";
+}
+
 export class MaxDurationExceededError extends Error {
   constructor(
     public readonly maxDurationInSeconds: number,
diff --git a/packages/core/src/v3/realtimeStreams/streamsWriterV2.test.ts b/packages/core/src/v3/realtimeStreams/streamsWriterV2.test.ts
@@ -0,0 +1,150 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+
+import { ChatChunkTooLargeError, isChatChunkTooLargeError } from "../errors.js";
+
+const lastAckedPosition = vi.fn(() => undefined);
+
+const appendSession = vi.fn(async () => {
+  // A WritableStream that just consumes records — we never reach S2 because
+  // the size check fires upstream of this for the oversize case, but we still
+  // need a valid writable for the small-chunk path.
+  const writable = new WritableStream<unknown>({});
+  return {
+    writable,
+    lastAckedPosition,
+  };
+});
+
+vi.mock("@s2-dev/streamstore", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("@s2-dev/streamstore")>();
+  return {
+    ...actual,
+    S2: class FakeS2 {
+      basin() {
+        return {
+          stream: () => ({
+            appendSession,
+          }),
+        };
+      }
+    },
+  };
+});
+
+import { StreamsWriterV2 } from "./streamsWriterV2.js";
+
+afterEach(() => {
+  vi.clearAllMocks();
+});
+
+describe("StreamsWriterV2", () => {
+  it("rejects with ChatChunkTooLargeError when a single chunk exceeds the per-record cap", async () => {
+    const oversized = {
+      type: "tool-output-available",
+      output: { text: "x".repeat(2_000_000) },
+    };
+    const source = new ReadableStream<unknown>({
+      start(controller) {
+        controller.enqueue(oversized);
+        controller.close();
+      },
+    });
+
+    const writer = new StreamsWriterV2({
+      basin: "test",
+      stream: "test",
+      accessToken: "test",
+      source,
+    });
+
+    await expect(writer.wait()).rejects.toBeInstanceOf(ChatChunkTooLargeError);
+
+    let captured: unknown;
+    try {
+      await writer.wait();
+    } catch (err) {
+      captured = err;
+    }
+    expect(isChatChunkTooLargeError(captured)).toBe(true);
+    const e = captured as ChatChunkTooLargeError;
+    expect(e.chunkType).toBe("tool-output-available");
+    expect(e.chunkSize).toBeGreaterThan(1_000_000);
+    expect(e.maxSize).toBe(1024 * 1024 - 1024);
+    expect(e.message).toMatch(/tool-output-available/);
+    expect(e.message).toMatch(/chat\.agent chunk/);
+  });
+
+  it("uses chunk.kind when chunk.type is missing (ChatInputChunk-style)", async () => {
+    const oversized = {
+      kind: "action",
+      payload: "x".repeat(2_000_000),
+    };
+    const source = new ReadableStream<unknown>({
+      start(controller) {
+        controller.enqueue(oversized);
+        controller.close();
+      },
+    });
+
+    const writer = new StreamsWriterV2({
+      basin: "test",
+      stream: "test",
+      accessToken: "test",
+      source,
+    });
+
+    let captured: unknown;
+    try {
+      await writer.wait();
+    } catch (err) {
+      captured = err;
+    }
+    expect(isChatChunkTooLargeError(captured)).toBe(true);
+    expect((captured as ChatChunkTooLargeError).chunkType).toBe("action");
+  });
+
+  it("omits chunkType when chunk has no discriminant", async () => {
+    const oversized = "x".repeat(2_000_000);
+    const source = new ReadableStream<unknown>({
+      start(controller) {
+        controller.enqueue(oversized);
+        controller.close();
+      },
+    });
+
+    const writer = new StreamsWriterV2({
+      basin: "test",
+      stream: "test",
+      accessToken: "test",
+      source,
+    });
+
+    let captured: unknown;
+    try {
+      await writer.wait();
+    } catch (err) {
+      captured = err;
+    }
+    expect(isChatChunkTooLargeError(captured)).toBe(true);
+    expect((captured as ChatChunkTooLargeError).chunkType).toBeUndefined();
+  });
+
+  it("does not reject for chunks under the cap", async () => {
+    const small = { type: "text-delta", delta: "hello" };
+    const source = new ReadableStream<unknown>({
+      start(controller) {
+        controller.enqueue(small);
+        controller.close();
+      },
+    });
+
+    const writer = new StreamsWriterV2({
+      basin: "test",
+      stream: "test",
+      accessToken: "test",
+      source,
+    });
+
+    await expect(writer.wait()).resolves.toBeDefined();
+  });
+});
diff --git a/packages/core/src/v3/realtimeStreams/streamsWriterV2.ts b/packages/core/src/v3/realtimeStreams/streamsWriterV2.ts
@@ -1,7 +1,16 @@
 import { S2, AppendRecord, BatchTransform } from "@s2-dev/streamstore";
+import { ChatChunkTooLargeError } from "../errors.js";
 import { StreamsWriter, StreamWriteResult } from "./types.js";
 import { nanoid } from "nanoid";
 
+// S2 caps a single record at 1 MiB of metered bytes (body + headers + 8 byte
+// overhead). We give ourselves ~1 KiB of headroom for the JSON envelope and
+// metering bytes so the check fires before the SDK's internal `BatchTransform`
+// rejects the record with an opaque `S2Error`.
+const RECORD_BODY_MAX_BYTES = 1024 * 1024 - 1024;
+
+const utf8Encoder = new TextEncoder();
+
 export type StreamsWriterV2Options<T = any> = {
   basin: string;
   stream: string;
@@ -152,8 +161,16 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
                 controller.error(new Error("Stream aborted"));
                 return;
               }
-              // Convert each chunk to JSON string and wrap in AppendRecord
-              controller.enqueue(AppendRecord.string({ body: JSON.stringify({ data: chunk, id: nanoid(7) }) }));
+              const body = JSON.stringify({ data: chunk, id: nanoid(7) });
+              const size = utf8Encoder.encode(body).length;
+              if (size > RECORD_BODY_MAX_BYTES) {
+                const chunkType = extractChunkType(chunk);
+                controller.error(
+                  new ChatChunkTooLargeError(size, RECORD_BODY_MAX_BYTES, chunkType)
+                );
+                return;
+              }
+              controller.enqueue(AppendRecord.string({ body }));
             },
           })
         )
@@ -227,3 +244,17 @@ function safeReleaseLock(reader: ReadableStreamDefaultReader<any>) {
     reader.releaseLock();
   } catch (error) {}
 }
+
+// chat.agent emits two chunk shapes through this writer:
+//   - UIMessageChunks + custom data parts: `{ type: "tool-output-available" | "data-..." | ... }`
+//   - ChatInputChunks (mostly seen on `.in`, but reused as the discriminant
+//     elsewhere): `{ kind: "message" | "stop" | "action" }`
+// Surfacing whichever discriminant exists turns "chunk too large" into
+// "tool-output-available chunk too large", which is what users actually need.
+function extractChunkType(chunk: unknown): string | undefined {
+  if (!chunk || typeof chunk !== "object") return undefined;
+  const c = chunk as { type?: unknown; kind?: unknown };
+  if (typeof c.type === "string") return c.type;
+  if (typeof c.kind === "string") return c.kind;
+  return undefined;
+}
diff --git a/packages/trigger-sdk/src/v3/index.ts b/packages/trigger-sdk/src/v3/index.ts
@@ -40,6 +40,8 @@ export {
   AbortTaskRunError,
   OutOfMemoryError,
   CompleteTaskWithOutput,
+  ChatChunkTooLargeError,
+  isChatChunkTooLargeError,
   logger,
   type LogLevel,
 } from "@trigger.dev/core/v3";
